diff --git a/.github/workflows/_ci-rust.yml b/.github/workflows/_ci-rust.yml index e25639d3ac7..a09ca691b71 100644 --- a/.github/workflows/_ci-rust.yml +++ b/.github/workflows/_ci-rust.yml @@ -66,6 +66,12 @@ jobs: tool: cargo-nextest@0.9.136 - name: Unit tests run: just test-unit + # Relay gateway (Bishop's keyless relay forwarder): infra-free tests + # against a loopback mock relay and Worker. Kept out of the Justfile so + # gateway PRs do not select the desktop suite. nextest gives each test + # its own process, which tests/logs.rs needs for its global subscriber. + - name: Unit tests (buzz-relay-gateway) + run: cargo nextest run -p buzz-relay-gateway server-cross-compile: name: Server Cross-Compile diff --git a/Cargo.lock b/Cargo.lock index de3f4d5c092..69f22573556 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1505,6 +1505,32 @@ dependencies = [ "uuid", ] +[[package]] +name = "buzz-relay-gateway" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "base64 0.22.1", + "buzz-auth", + "futures-util", + "hex", + "nostr 0.44.7", + "reqwest 0.13.4", + "rustls", + "serde", + "serde_json", + "sha2 0.11.0", + "thiserror 2.0.18", + "tokio", + "tokio-tungstenite 0.29.0", + "tokio-util", + "tracing", + "tracing-subscriber", + "url", + "uuid", +] + [[package]] name = "buzz-relay-mesh" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index df77ab7cb3c..51f1db14a9f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,6 +32,7 @@ members = [ "crates/buzz-dev-mcp", "crates/buzz-voice", "crates/buzz-audio-client", + "crates/buzz-relay-gateway", "crates/buzz-backend-kubernetes", "crates/buzz-datastore-tracing", "examples/countdown-bot", diff --git a/Dockerfile.relay-gateway b/Dockerfile.relay-gateway new file mode 100644 index 00000000000..ae29eedb94c --- /dev/null +++ b/Dockerfile.relay-gateway @@ -0,0 +1,28 @@ +# syntax=docker/dockerfile:1.7 +# Buzz relay gateway: a keyless forwarder between one Worker link and one Buzz relay. +# The build context is the repo root (cargo needs the workspace manifests and Cargo.lock): +# docker build -f Dockerfile.relay-gateway --platform linux/amd64 -t buzz-relay-gateway:dev . +# Base images pinned by digest (first build, 27 Oct 2026). +FROM --platform=linux/amd64 rust:1.95.0-alpine@sha256:606fd313a0f49743ee2a7bd49a0914bab7deedb12791f3a846a34a4711db7ed2 AS builder +# build-base for ring and aws-lc-sys (C); cmake in case aws-lc-sys picks its cmake builder. +RUN apk add --no-cache build-base cmake +WORKDIR /build +COPY . . +# The image already has 1.95.0; this skips the components rust-toolchain.toml would install. +ENV RUSTUP_TOOLCHAIN=1.95.0 +RUN --mount=type=cache,target=/usr/local/cargo/registry \ + --mount=type=cache,target=/usr/local/cargo/git \ + --mount=type=cache,target=/build/target \ + cargo build --release --locked -p buzz-relay-gateway --bin buzz-relay-gateway \ + && install -D -m 0755 target/release/buzz-relay-gateway /out/buzz-relay-gateway \ + && strip /out/buzz-relay-gateway + +FROM gcr.io/distroless/static-debian12:nonroot@sha256:afa5c872c891853ca7fcf1f12c3edb23f7eeef36189728842dd51042ff57f7ab +LABEL org.opencontainers.image.title="Buzz relay gateway" \ + org.opencontainers.image.description="Keyless forwarder from a Worker link to one Buzz relay" \ + org.opencontainers.image.licenses="Apache-2.0" +COPY --from=builder /out/buzz-relay-gateway /usr/local/bin/buzz-relay-gateway +ENV BUZZ_RELAY_GATEWAY_BIND_ADDR=0.0.0.0:8080 RUST_LOG=info +EXPOSE 8080 +USER nonroot:nonroot +ENTRYPOINT ["/usr/local/bin/buzz-relay-gateway"] diff --git a/crates/buzz-relay-gateway/Cargo.toml b/crates/buzz-relay-gateway/Cargo.toml new file mode 100644 index 00000000000..33243ae84cc --- /dev/null +++ b/crates/buzz-relay-gateway/Cargo.toml @@ -0,0 +1,42 @@ +[package] +name = "buzz-relay-gateway" +description = "Keyless forwarder between one Worker link and one Buzz relay" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true + +[lib] +name = "buzz_relay_gateway" +path = "src/lib.rs" + +[[bin]] +name = "buzz-relay-gateway" +path = "src/main.rs" + +[dependencies] +anyhow = { workspace = true } +axum = { workspace = true } +base64 = { workspace = true } +buzz-auth = { workspace = true } +futures-util = { workspace = true } +nostr = { workspace = true } +reqwest = { workspace = true } +# Explicit ring provider, installed in main(): a workspace-wide build unifies +# ring and aws-lc-rs, and rustls then panics on the first wss:// connect. +rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +tokio-tungstenite = { workspace = true } +tokio-util = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +url = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +hex = { workspace = true } +sha2 = { workspace = true } diff --git a/crates/buzz-relay-gateway/README.md b/crates/buzz-relay-gateway/README.md new file mode 100644 index 00000000000..7e03649a3db --- /dev/null +++ b/crates/buzz-relay-gateway/README.md @@ -0,0 +1,61 @@ +# Buzz relay gateway + +`buzz-relay-gateway` is a keyless forwarder between one Worker link and one +Buzz relay. It exists because a Cloudflare Worker cannot reach a relay that +sits behind Cloudflare itself (error 526), while a Cloudflare Container can. +The Worker keeps every key and signs every event; the gateway only checks the +link bearer and moves bytes. + +## Trust boundary + +- The gateway holds no secret. Its environment is four public values. +- The Worker opens `GET /v1/link` with `Authorization: Nostr `: + a kind 27235 event signed by a dedicated gateway-auth key (never a relay + member), `u` = the configured link URL, `method` = `GET`, within ±60 s, with + one `nonce`. Each bearer works once, and a link lives at most one hour. +- The gateway authenticates the Worker by signature. The Worker authenticates + the gateway only by transport (the Durable Object to container channel, or + TLS on a VM). +- Only one relay origin is reachable, through an exact route table: + `POST /query`, `/events`, `/count`, `GET /` (NIP-11) and the WebSocket on + `/`. Request and response headers are allowlisted, redirects are not + followed, nothing is retried, and bodies are never re-encoded, so NIP-98 + `payload` tags still verify at the relay. +- NIP-42 is transparent: the relay's `AUTH` challenge reaches the Worker as an + ordinary `ws_msg`, and the Worker signs the reply. +- Logs carry ids, status codes, close codes and timings. They never carry + bearers, bodies, relay frames or close-reason text. + +The frames are in `src/frame.rs`, the route table in `src/policy.rs`, the +limits in `src/config.rs`, and the link close codes in `src/link.rs`. + +## Environment + +| Variable | Required | Value | +|---|---|---| +| `BUZZ_RELAY_GATEWAY_BIND_ADDR` | no | default `127.0.0.1:8080`; the image sets `0.0.0.0:8080` | +| `BUZZ_RELAY_GATEWAY_RELAY_URL` | yes | `wss://host[:port]`; `ws://` only on loopback | +| `BUZZ_RELAY_GATEWAY_LINK_URL` | yes | the `u` every link bearer must carry | +| `BUZZ_RELAY_GATEWAY_LINK_PUBKEY` | yes | the gateway-auth public key, 64 lowercase hex | +| `RUST_LOG` | no | `info`; `tungstenite` and `tokio_tungstenite` stay off at any level, since at trace they print whole frames | + +`GET /healthz` answers `ok`, or 503 once shutdown has begun. SIGTERM or +Ctrl-C drains in-flight requests for up to 10 s, closes every relay socket, +then closes each link with 1012. + +## Testing + +``` +cargo test -p buzz-relay-gateway +cargo clippy -p buzz-relay-gateway --all-targets -- -D warnings +``` + +The tests run offline against an in-process mock relay (HTTP routes and a +NIP-42 socket) and a mock Worker. CI runs them in the Unit Tests job (`.github/workflows/_ci-rust.yml`). + +The image builds from the repo root. Pushing it is a deployment step, not a +build step: + +``` +docker build -f Dockerfile.relay-gateway --platform linux/amd64 -t buzz-relay-gateway:dev . +``` diff --git a/crates/buzz-relay-gateway/src/auth.rs b/crates/buzz-relay-gateway/src/auth.rs new file mode 100644 index 00000000000..ae8206ca3bb --- /dev/null +++ b/crates/buzz-relay-gateway/src/auth.rs @@ -0,0 +1,289 @@ +//! The link bearer: a single-use NIP-98 event signed by the gateway-auth key. + +use std::{ + collections::HashMap, + sync::{Mutex, PoisonError}, + time::{Duration, Instant}, +}; + +use axum::http::{header::AUTHORIZATION, HeaderMap}; +use base64::{engine::general_purpose::STANDARD, Engine as _}; +use buzz_auth::{nip98_replay::DEFAULT_REPLAY_TTL_SECS, verify_nip98_event}; +use nostr::{Event, EventId, PublicKey}; + +/// Why a bearer was refused. Only [`Reject::code`] is logged: the underlying +/// error text quotes bearer content. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Reject { + /// No `Authorization` header. + Missing, + /// Repeated, oversized, not `Nostr `, or not UTF-8. + Malformed, + /// Failed NIP-98: kind, signature, time window, `u` or `method`. + Invalid, + /// A valid signature by a key other than the pinned one. + WrongKey, + /// Not exactly one `nonce` tag of 16 to 128 characters. + NoNonce, + /// The event id was already used. + Replayed, + /// The replay cache is full; the gateway fails closed. + CacheFull, +} + +impl Reject { + /// The fixed reason code for logs. + pub fn code(self) -> &'static str { + match self { + Self::Missing => "missing", + Self::Malformed => "malformed", + Self::Invalid => "invalid", + Self::WrongKey => "wrong_key", + Self::NoNonce => "no_nonce", + Self::Replayed => "replayed", + Self::CacheFull => "cache_full", + } + } +} + +/// Checks the link bearer on an upgrade request and returns its event id and +/// `created_at` (unix seconds), which the caller then claims in the [`ReplayCache`]. +pub fn verify_link_bearer( + headers: &HeaderMap, + link_url: &str, + pubkey: &PublicKey, + max_len: usize, +) -> Result<(EventId, u64), Reject> { + let mut values = headers.get_all(AUTHORIZATION).iter(); + let (value, None) = (values.next().ok_or(Reject::Missing)?, values.next()) else { + return Err(Reject::Malformed); + }; + let json = (value.len() <= max_len) + .then(|| value.to_str().ok()?.strip_prefix("Nostr ")) + .flatten() + .and_then(|b64| STANDARD.decode(b64).ok()) + .and_then(|bytes| String::from_utf8(bytes).ok()) + .ok_or(Reject::Malformed)?; + verify_nip98_event(&json, link_url, "GET", None).map_err(|_| Reject::Invalid)?; + let event: Event = serde_json::from_str(&json).map_err(|_| Reject::Invalid)?; + if event.pubkey != *pubkey { + return Err(Reject::WrongKey); + } + let mut nonces = event + .tags + .iter() + .map(|t| t.as_slice()) + .filter(|t| t.first().is_some_and(|n| n == "nonce")); + match (nonces.next(), nonces.next()) { + (Some([_, n, ..]), None) if (16..=128).contains(&n.chars().count()) => { + Ok((event.id, event.created_at.as_secs())) + } + _ => Err(Reject::NoNonce), + } +} + +/// How long a claimed id is kept: the relay's replay TTL (twice the ±60 s +/// window) plus 2 s, because the window is inclusive and counts whole seconds, +/// so a bearer dated 60 s ahead stays valid for up to 121 s. +const REPLAY_TTL: Duration = Duration::from_secs(DEFAULT_REPLAY_TTL_SECS + 2); + +/// `verify_nip98_event`'s ±60 s window on `created_at` (the relay's TTL is twice it). +const WINDOW_SECS: u64 = DEFAULT_REPLAY_TTL_SECS / 2; + +/// Event ids of accepted bearers, each kept for `REPLAY_TTL` of monotonic time +/// and, if the wall clock steps back, for as long as the window could still +/// accept its `created_at`. A restart empties it. +pub struct ReplayCache { + cap: usize, + seen: Mutex>, +} + +impl ReplayCache { + /// A cache that holds at most `cap` live ids. + pub fn new(cap: usize) -> Self { + Self { + cap, + seen: Mutex::new(HashMap::new()), + } + } + + /// Claims a bearer `(id, created_at)` once, `now` being monotonic and `wall` + /// unix seconds. Refuses a replay, and refuses every id while full. + pub fn claim( + &self, + (id, created_at): (EventId, u64), + now: Instant, + wall: u64, + ) -> Result<(), Reject> { + let mut seen = self.seen.lock().unwrap_or_else(PoisonError::into_inner); + seen.retain(|_, (at, created)| { + now.saturating_duration_since(*at) < REPLAY_TTL + || wall <= created.saturating_add(WINDOW_SECS + 1) + }); + if seen.contains_key(&id) { + return Err(Reject::Replayed); + } + if seen.len() >= self.cap { + return Err(Reject::CacheFull); + } + seen.insert(id, (now, created_at)); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::http::HeaderValue; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + const URL: &str = "https://gateway.bishop.invalid/v1/link"; + const NONCE: &str = "6f1c2c1e-4b9a-4c71-9a55-0f1d8c3a2b10"; + + fn event(keys: &Keys, kind: u16, tags: &[&[&str]], age: i64) -> Event { + let tags = tags.iter().map(|t| Tag::parse(t.iter().copied()).unwrap()); + let at = Timestamp::from((Timestamp::now().as_secs() as i64 - age) as u64); + EventBuilder::new(Kind::from(kind), "") + .tags(tags) + .custom_created_at(at) + .sign_with_keys(keys) + .unwrap() + } + + fn header(event: &Event) -> String { + format!( + "Nostr {}", + STANDARD.encode(serde_json::to_string(event).unwrap()) + ) + } + + fn good(keys: &Keys) -> String { + let nonce = uuid::Uuid::new_v4().to_string(); + header(&event( + keys, + 27235, + &[&["u", URL], &["method", "GET"], &["nonce", &nonce]], + 0, + )) + } + + fn check(keys: &Keys, values: &[&str]) -> Result<(EventId, u64), Reject> { + let mut headers = HeaderMap::new(); + for v in values { + headers.append(AUTHORIZATION, HeaderValue::from_str(v).unwrap()); + } + verify_link_bearer(&headers, URL, &keys.public_key(), 8 << 10) + } + + #[test] + fn accepts_a_good_bearer_and_url_normalisation() { + let k = Keys::generate(); + assert!(check(&k, &[&good(&k)]).is_ok()); + for u in [ + "https://gateway.bishop.invalid/v1/link/", + "HTTPS://Gateway.Bishop.Invalid/v1/link", + ] { + let b = header(&event( + &k, + 27235, + &[&["u", u], &["method", "get"], &["nonce", NONCE]], + 0, + )); + assert!(check(&k, &[&b]).is_ok(), "{u}"); + } + } + + #[test] + fn refuses_bad_bearers() { + let k = Keys::generate(); + let with = |kind, tags: &[&[&str]], age| header(&event(&k, kind, tags, age)); + let (u, get, nonce) = ( + &["u", URL][..], + &["method", "GET"][..], + &["nonce", NONCE][..], + ); + let mut tampered: serde_json::Value = + serde_json::to_value(event(&k, 27235, &[u, get, nonce], 0)).unwrap(); + tampered["content"] = "x".into(); + let tampered = format!("Nostr {}", STANDARD.encode(tampered.to_string())); + let cases = [ + (good(&Keys::generate()), Reject::WrongKey), + (with(27235, &[u, get, nonce], 120), Reject::Invalid), + (with(27235, &[u, get, nonce], -120), Reject::Invalid), + ( + with( + 27235, + &[&["u", "https://other.invalid/v1/link"], get, nonce], + 0, + ), + Reject::Invalid, + ), + ( + with(27235, &[u, &["method", "POST"], nonce], 0), + Reject::Invalid, + ), + (with(22242, &[u, get, nonce], 0), Reject::Invalid), + (tampered, Reject::Invalid), + (with(27235, &[u, get], 0), Reject::NoNonce), + ( + with( + 27235, + &[u, get, nonce, &["nonce", "a-second-nonce-value"]], + 0, + ), + Reject::NoNonce, + ), + ( + with(27235, &[u, get, &["nonce", "short"]], 0), + Reject::NoNonce, + ), + (good(&k).replacen("Nostr ", "Bearer ", 1), Reject::Malformed), + ("Nostr !!!".to_owned(), Reject::Malformed), + ( + format!("Nostr {}", STANDARD.encode([0xff, 0xfe])), + Reject::Malformed, + ), + (format!("Nostr {}", "A".repeat(8 << 10)), Reject::Malformed), + ]; + for (i, (bearer, want)) in cases.iter().enumerate() { + assert_eq!(check(&k, &[bearer]), Err(*want), "case {i}"); + } + assert_eq!(check(&k, &[]), Err(Reject::Missing)); + assert_eq!(check(&k, &[&good(&k), &good(&k)]), Err(Reject::Malformed)); + } + + #[test] + fn replay_cache_is_single_use_prunes_and_fails_closed() { + let k = Keys::generate(); + let (a, b) = ( + check(&k, &[&good(&k)]).unwrap(), + check(&k, &[&good(&k)]).unwrap(), + ); + let cache = ReplayCache::new(1); + let (now, wall) = (Instant::now(), a.1); + assert_eq!(cache.claim(a, now, wall), Ok(())); + assert_eq!(cache.claim(a, now, wall), Err(Reject::Replayed)); + assert_eq!(cache.claim(b, now, wall), Err(Reject::CacheFull)); + // Still held when the relay's own 120 s would have let it go. + let relay_ttl = now + Duration::from_secs(DEFAULT_REPLAY_TTL_SECS); + assert_eq!(cache.claim(a, relay_ttl, wall + 120), Err(Reject::Replayed)); + assert_eq!(cache.claim(b, now + REPLAY_TTL, wall + 122), Ok(())); + } + + #[test] + fn a_wall_clock_step_back_keeps_the_id_while_the_window_accepts_it() { + let k = Keys::generate(); + let a = check(&k, &[&good(&k)]).unwrap(); + let cache = ReplayCache::new(4); + let now = Instant::now(); + assert_eq!(cache.claim(a, now, a.1), Ok(())); + // Monotonic time is past REPLAY_TTL, but the wall clock reads + // created_at + 60 s, which the verifier still accepts. + let later = now + REPLAY_TTL; + assert_eq!( + cache.claim(a, later, a.1 + WINDOW_SECS), + Err(Reject::Replayed) + ); + assert_eq!(cache.claim(a, later, a.1 + WINDOW_SECS + 2), Ok(())); + } +} diff --git a/crates/buzz-relay-gateway/src/config.rs b/crates/buzz-relay-gateway/src/config.rs new file mode 100644 index 00000000000..57f4df543e7 --- /dev/null +++ b/crates/buzz-relay-gateway/src/config.rs @@ -0,0 +1,234 @@ +//! Environment configuration (all public values) and the fixed limits. + +use std::{net::SocketAddr, time::Duration}; + +use nostr::PublicKey; +use url::Url; + +const BIND: &str = "BUZZ_RELAY_GATEWAY_BIND_ADDR"; +const RELAY: &str = "BUZZ_RELAY_GATEWAY_RELAY_URL"; +const LINK_URL: &str = "BUZZ_RELAY_GATEWAY_LINK_URL"; +const LINK_PUBKEY: &str = "BUZZ_RELAY_GATEWAY_LINK_PUBKEY"; + +/// The gateway's settings. None of them is a secret. +#[derive(Clone, Debug)] +pub struct Config { + /// Listen address; defaults to loopback. + pub bind_addr: SocketAddr, + /// The one relay the gateway reaches, `wss://host[:port]/`. + pub relay_ws: Url, + /// The same relay's HTTP origin, `https://host[:port]/`. + pub relay_http: Url, + /// The `u` tag every link bearer must carry. + pub link_url: String, + /// The gateway-auth public key that signs link bearers. + pub link_pubkey: PublicKey, +} + +/// A missing or invalid variable. It names the variable, never its value. +#[derive(Debug, thiserror::Error)] +pub enum ConfigError { + /// A required variable is not set. + #[error("{0} is not set")] + Missing(&'static str), + /// A variable is set but does not parse or is not allowed. + #[error("{0} is invalid")] + Invalid(&'static str), +} + +impl Config { + /// Reads `BUZZ_RELAY_GATEWAY_*` from the process environment. + pub fn from_env() -> Result { + Self::from_lookup(|name| std::env::var(name).ok()) + } + + /// Reads the settings through `get`, so tests need no process environment. + pub fn from_lookup(get: impl Fn(&str) -> Option) -> Result { + let required = |name: &'static str| get(name).ok_or(ConfigError::Missing(name)); + let bind_addr = get(BIND) + .as_deref() + .unwrap_or("127.0.0.1:8080") + .parse() + .map_err(|_| ConfigError::Invalid(BIND))?; + let (relay_ws, relay_http) = + relay_urls(&required(RELAY)?).ok_or(ConfigError::Invalid(RELAY))?; + let link_url = required(LINK_URL)?; + Url::parse(&link_url).map_err(|_| ConfigError::Invalid(LINK_URL))?; + let hex = required(LINK_PUBKEY)?; + let lower_hex = + hex.len() == 64 && hex.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')); + let link_pubkey = PublicKey::from_hex(&hex) + .ok() + .filter(|_| lower_hex) + .ok_or(ConfigError::Invalid(LINK_PUBKEY))?; + Ok(Self { + bind_addr, + relay_ws, + relay_http, + link_url, + link_pubkey, + }) + } +} + +/// `wss://host[:port]` with no userinfo, path, query or fragment; `ws://` only on loopback. +fn relay_urls(raw: &str) -> Option<(Url, Url)> { + let ws = Url::parse(raw).ok()?; + let loopback = matches!(ws.host_str()?, "127.0.0.1" | "[::1]" | "localhost"); + let scheme = match ws.scheme() { + "wss" => "https", + "ws" if loopback => "http", + _ => return None, + }; + let bare = ws.username().is_empty() + && ws.password().is_none() + && ws.path() == "/" + && ws.query().is_none() + && ws.fragment().is_none(); + let mut http = ws.clone(); + http.set_scheme(scheme).ok()?; + bare.then_some((ws, http)) +} + +/// Fixed limits and timeouts (design §5). Not configurable by env; tests pass smaller values. +#[derive(Clone, Debug)] +pub struct Limits { + /// Links open at once; the next upgrade gets 503. + pub max_links: usize, + /// Longest `Authorization` value, on the link bearer and on forwarded requests. + pub max_auth_header: usize, + /// Bearer ids remembered for replay protection; when full, new bearers get 503. + pub replay_entries: usize, + /// Largest Worker frame; a bigger one closes the link with 1009. + pub max_frame: usize, + /// Largest decoded request body (the relay's own API limit). + pub max_request_body: usize, + /// Largest relay response body; the read stops there. + pub max_response_body: usize, + /// Largest `ws_send` text (the relay's frame limit). + pub max_ws_text: usize, + /// Largest frame the gateway accepts from the relay. + pub max_relay_frame: usize, + /// HTTP requests in flight per link. + pub max_http_inflight: usize, + /// Relay sockets per link. + pub max_sockets: usize, + /// Frames queued for the Worker; a full queue closes the link with 4003. + pub out_frames: usize, + /// Serialized bytes queued for the Worker, with the same rule. + pub out_bytes: usize, + /// Worker frames queued for one relay socket; a full queue closes that socket. + pub relay_queue: usize, + /// How long a link lives before it closes with 4001. + pub link_lifetime: Duration, + /// How long before that the gateway sends `rotate`. + pub rotate_before: Duration, + /// Gateway `ping` interval. + pub ping_every: Duration, + /// A link with no Worker frame for this long closes with 4002. + pub idle: Duration, + /// Reaching the relay: TCP, TLS and, for sockets, the upgrade. + pub connect_timeout: Duration, + /// One whole HTTP exchange with the relay. + pub request_timeout: Duration, + /// A relay socket with no frame (pings included) for this long is closed. + pub relay_silence: Duration, + /// How long shutdown waits for in-flight HTTP requests. + pub shutdown_grace: Duration, +} + +impl Default for Limits { + fn default() -> Self { + Self { + max_links: 4, + max_auth_header: 8 << 10, + replay_entries: 1024, + max_frame: 2 << 20, + max_request_body: 1 << 20, + max_response_body: 4 << 20, + max_ws_text: 512 << 10, + max_relay_frame: 1 << 20, + max_http_inflight: 16, + max_sockets: 8, + out_frames: 1024, + out_bytes: 16 << 20, + relay_queue: 64, + link_lifetime: Duration::from_secs(3600), + rotate_before: Duration::from_secs(300), + ping_every: Duration::from_secs(20), + idle: Duration::from_secs(60), + connect_timeout: Duration::from_secs(10), + request_timeout: Duration::from_secs(30), + relay_silence: Duration::from_secs(90), + shutdown_grace: Duration::from_secs(10), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // The secp256k1 generator x: a valid key whose secret (1) is public anyway. + const PK: &str = "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"; + + fn load(relay: &str, pubkey: &str) -> Result { + Config::from_lookup(|name| match name { + RELAY => Some(relay.to_owned()), + LINK_URL => Some("https://gateway.bishop.invalid/v1/link".to_owned()), + LINK_PUBKEY => Some(pubkey.to_owned()), + _ => None, + }) + } + + #[test] + fn accepts_wss_and_loopback_ws() { + let c = load("wss://relay.example.com", PK).unwrap(); + assert_eq!(c.relay_ws.as_str(), "wss://relay.example.com/"); + assert_eq!(c.relay_http.as_str(), "https://relay.example.com/"); + assert_eq!(c.bind_addr.to_string(), "127.0.0.1:8080"); + for ok in [ + "ws://127.0.0.1:7000", + "ws://localhost:7000/", + "ws://[::1]:7000", + "wss://r.example:444", + ] { + assert!(load(ok, PK).is_ok(), "{ok}"); + } + assert_eq!( + load("ws://127.0.0.1:7000", PK).unwrap().relay_http.as_str(), + "http://127.0.0.1:7000/" + ); + } + + #[test] + fn refuses_other_relay_urls() { + for bad in [ + "ws://example.com", + "wss://relay.example.com/path", + "wss://relay.example.com/?q=1", + "wss://relay.example.com/#f", + "wss://user:pw@relay.example.com", + "https://relay.example.com", + "http://127.0.0.1:7000", + "relay.example.com", + ] { + assert!( + matches!(load(bad, PK), Err(ConfigError::Invalid(RELAY))), + "{bad}" + ); + } + } + + #[test] + fn refuses_bad_pubkey_and_names_missing_variables() { + for bad in [&PK[..63], &PK.to_uppercase(), &format!("{}zz", &PK[..62])] { + assert!(matches!( + load("wss://r.example", bad), + Err(ConfigError::Invalid(LINK_PUBKEY)) + )); + } + let err = Config::from_lookup(|_| None).unwrap_err(); + assert_eq!(err.to_string(), "BUZZ_RELAY_GATEWAY_RELAY_URL is not set"); + } +} diff --git a/crates/buzz-relay-gateway/src/frame.rs b/crates/buzz-relay-gateway/src/frame.rs new file mode 100644 index 00000000000..0486c19ef05 --- /dev/null +++ b/crates/buzz-relay-gateway/src/frame.rs @@ -0,0 +1,411 @@ +//! Link frames, version 1 (design §3): one JSON object per WebSocket text frame. +//! +//! The frame types that carry headers, bodies or relay text do not derive +//! `Debug`, so none of them can reach a log line by accident. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; +use serde_json::{value::RawValue, Number}; + +use crate::policy::Refused; + +/// A Worker-chosen request or socket id: 1 to 64 of `[A-Za-z0-9_-]`, so it is safe to log. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Deserialize, Serialize)] +#[serde(try_from = "String")] +pub struct Id(String); + +impl TryFrom for Id { + type Error = &'static str; + + fn try_from(s: String) -> Result { + let valid = (1..=64).contains(&s.len()) + && s.bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-'); + valid.then_some(Self(s)).ok_or("invalid id") + } +} + +impl Id { + /// The id, for logs. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// An `http` frame: one request to forward to the relay. +#[derive(Deserialize)] +pub struct HttpFrame { + /// Unique among this link's requests in flight. + pub id: Id, + /// `GET` or `POST`, matched exactly by [`crate::policy::http_route`]. + pub method: String, + /// Matched exactly by [`crate::policy::http_route`]. + pub path: String, + /// The raw `headers` object, checked by [`crate::policy::request_headers`]. + pub headers: Option>, + /// The request body, standard base64 (POST only). + pub body_b64: Option, +} + +/// A frame from the Worker. +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Inbound { + /// Forward one HTTP request. Parsed on its own: `RawValue` does not + /// survive serde's internally tagged enums. + #[serde(skip)] + Http(HttpFrame), + /// Open a relay socket on `path`. + WsOpen { sid: Id, path: String }, + /// Send `text` on a relay socket, as is. + WsSend { sid: Id, text: String }, + /// Close a relay socket. + WsClose { + sid: Id, + code: Option, + reason: Option, + }, + /// Liveness; answered with a `pong` carrying the same `ts`. + Ping { ts: Option }, + /// The answer to a gateway `ping`. + Pong, + /// A `type` this version does not know; the link stays open. + #[serde(other)] + Unknown, +} + +/// Not JSON, no `type`, or a known type with a missing or mistyped field. +#[derive(Debug, PartialEq, Eq)] +pub struct BadFrame; + +#[derive(Deserialize)] +struct TypeOnly { + #[serde(rename = "type")] + kind: String, +} + +impl Inbound { + /// Parses one text frame. Each type reads only its own fields, so a field + /// that belongs to another type is ignored like any unknown field. + pub fn parse(text: &str) -> Result { + let TypeOnly { kind } = serde_json::from_str(text).map_err(|_| BadFrame)?; + let frame = if kind == "http" { + serde_json::from_str(text).map(Self::Http) + } else { + serde_json::from_str(text) + }; + frame.map_err(|_| BadFrame) + } +} + +/// Which side ended a relay socket. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Side { + /// The relay closed it or the connection dropped. + Relay, + /// The gateway closed it (limits, link end, shutdown, failed open). + Gateway, + /// The Worker asked with `ws_close`. + Worker, +} + +/// The limits announced in `hello`. +#[derive(Clone, Copy, Debug, Serialize)] +pub struct HelloLimits { + pub max_request_body: usize, + pub max_response_body: usize, + pub max_ws_text: usize, + pub max_http_inflight: usize, + pub max_sockets: usize, +} + +/// A frame to the Worker. +#[derive(Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Out { + /// The first frame on every link. + Hello { + v: u8, + link_id: String, + expires_at: u64, + rotate_at: u64, + limits: HelloLimits, + }, + /// The relay's answer, whatever its status. + HttpRes { + id: Id, + status: u16, + headers: BTreeMap<&'static str, String>, + body_b64: String, + ms: u64, + }, + /// The relay accepted the socket upgrade. + WsOpened { sid: Id, ms: u64 }, + /// One text frame from the relay, as is. + WsMsg { sid: Id, text: String }, + /// The end of a socket, sent exactly once per `sid`. + WsClosed { + sid: Id, + by: Side, + code: Option, + reason: String, + #[serde(skip_serializing_if = "Option::is_none")] + http_status: Option, + }, + /// A frame the gateway refused or could not perform. + Error { + #[serde(skip_serializing_if = "Option::is_none")] + id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + sid: Option, + code: ErrorCode, + message: &'static str, + }, + /// The link closes with 4001 at `close_at` (unix seconds). + Rotate { close_at: u64 }, + /// Liveness; the Worker answers `pong`. + Ping { ts: Number }, + /// The answer to a Worker `ping`. + Pong { + #[serde(skip_serializing_if = "Option::is_none")] + ts: Option, + }, +} + +impl Out { + /// An `error` frame with the code's fixed message. + pub fn error(code: ErrorCode, id: Option, sid: Option) -> Self { + Self::Error { + id, + sid, + code, + message: code.message(), + } + } +} + +/// `error.code`. Each code has a fixed message that never quotes frame content. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ErrorCode { + BadFrame, + UnknownType, + NotAllowed, + NotSupported, + BadBase64, + BodyTooLarge, + FrameTooLarge, + ResponseTooLarge, + DuplicateId, + UnknownSocket, + Busy, + UpstreamConnect, + UpstreamTimeout, + ShuttingDown, +} + +impl ErrorCode { + /// The fixed English phrase for `message`. + pub fn message(self) -> &'static str { + match self { + Self::BadFrame => "malformed frame", + Self::UnknownType => "unknown frame type", + Self::NotAllowed => "not allowed", + Self::NotSupported => "not supported yet", + Self::BadBase64 => "body_b64 is not valid base64", + Self::BodyTooLarge => "request body too large", + Self::FrameTooLarge => "text frame too large", + Self::ResponseTooLarge => "relay response too large", + Self::DuplicateId => "id already in use", + Self::UnknownSocket => "no such socket", + Self::Busy => "too many requests or sockets in flight", + Self::UpstreamConnect => "could not reach the relay", + Self::UpstreamTimeout => "the relay did not answer in time", + Self::ShuttingDown => "gateway shutting down", + } + } +} + +impl From for ErrorCode { + fn from(refused: Refused) -> Self { + match refused { + Refused::NotAllowed => Self::NotAllowed, + Refused::NotSupported => Self::NotSupported, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn id(s: &str) -> Id { + Id::try_from(s.to_owned()).unwrap() + } + + #[test] + fn parses_every_inbound_type() { + let http = r#"{"type":"http","id":"q-1","method":"POST","path":"/query", + "headers":{"accept":"a","accept":"b"},"body_b64":"e30=","extra":1}"#; + let Ok(Inbound::Http(f)) = Inbound::parse(http) else { + panic!("http") + }; + assert_eq!( + (f.id, f.method.as_str(), f.path.as_str()), + (id("q-1"), "POST", "/query") + ); + // Kept raw, so the policy can still see the duplicated key. + assert_eq!(f.headers.unwrap().get(), r#"{"accept":"a","accept":"b"}"#); + assert_eq!(f.body_b64.as_deref(), Some("e30=")); + let open = Inbound::parse(r#"{"type":"ws_open","sid":"s1","path":"/"}"#); + assert!( + matches!(open, Ok(Inbound::WsOpen { sid, path }) if sid == id("s1") && path == "/") + ); + let send = Inbound::parse(r#"{"type":"ws_send","sid":"s1","text":"[\"AUTH\",{}]"}"#); + assert!(matches!(send, Ok(Inbound::WsSend { text, .. }) if text == r#"["AUTH",{}]"#)); + let close = Inbound::parse(r#"{"type":"ws_close","sid":"s1","code":4000,"reason":"bye"}"#); + assert!( + matches!(close, Ok(Inbound::WsClose { code: Some(4000), reason: Some(r), .. }) if r == "bye") + ); + let ping = Inbound::parse(r#"{"type":"ping","ts":1793149200123}"#); + assert!( + matches!(ping, Ok(Inbound::Ping { ts: Some(ts) }) if ts.to_string() == "1793149200123") + ); + assert!(matches!( + Inbound::parse(r#"{"type":"pong","ts":1}"#), + Ok(Inbound::Pong) + )); + assert!(matches!( + Inbound::parse(r#"{"type":"later"}"#), + Ok(Inbound::Unknown) + )); + assert!(matches!( + Inbound::parse(r#"{"type":"later","code":"x"}"#), + Ok(Inbound::Unknown) + )); + // Another frame type's field, of another type, is still an unknown field here. + for foreign in [ + r#"{"type":"ping","ts":1,"code":"later"}"#, + r#"{"type":"ping","ts":1,"id":7}"#, + r#"{"type":"pong","sid":[]}"#, + r#"{"type":"ws_send","sid":"s1","text":"x","path":5}"#, + r#"{"type":"http","id":"q","method":"GET","path":"/","ts":"x","text":1}"#, + ] { + assert!(Inbound::parse(foreign).is_ok(), "{foreign}"); + } + } + + #[test] + fn refuses_bad_frames_and_ids() { + for bad in [ + "not json", + "[]", + r#"{"id":"x"}"#, + r#"{"type":"http","id":"q","method":"POST"}"#, + r#"{"type":"ws_send","sid":"s1"}"#, + r#"{"type":"ws_open","sid":"has space","path":"/"}"#, + r#"{"type":"ws_open","sid":"","path":"/"}"#, + r#"{"type":"ping","ts":"soon"}"#, + r#"{"type":5}"#, + r#"{"type":"ws_open","sid":"s1","sid":"s2","path":"/"}"#, + ] { + assert!(Inbound::parse(bad).is_err(), "{bad}"); + } + let long = format!( + r#"{{"type":"ws_open","sid":"{}","path":"/"}}"#, + "a".repeat(65) + ); + assert_eq!(Inbound::parse(&long).err(), Some(BadFrame)); + assert!(Id::try_from("A-z_09".to_owned()).is_ok()); + assert!(Id::try_from("a".repeat(64)).is_ok()); + } + + #[test] + fn serializes_every_outbound_type() { + let limits = HelloLimits { + max_request_body: 1, + max_response_body: 2, + max_ws_text: 3, + max_http_inflight: 4, + max_sockets: 5, + }; + let headers = BTreeMap::from([("content-type", "application/json".to_owned())]); + let cases = [ + ( + Out::Hello { + v: 1, + link_id: "l".into(), + expires_at: 10, + rotate_at: 9, + limits, + }, + json!({"type":"hello","v":1,"link_id":"l","expires_at":10,"rotate_at":9,"limits": + {"max_request_body":1,"max_response_body":2,"max_ws_text":3,"max_http_inflight":4,"max_sockets":5}}), + ), + ( + Out::HttpRes { + id: id("q"), + status: 429, + headers, + body_b64: "e30=".into(), + ms: 7, + }, + json!({"type":"http_res","id":"q","status":429,"headers":{"content-type":"application/json"},"body_b64":"e30=","ms":7}), + ), + ( + Out::WsOpened { + sid: id("s"), + ms: 3, + }, + json!({"type":"ws_opened","sid":"s","ms":3}), + ), + ( + Out::WsMsg { + sid: id("s"), + text: "[1]".into(), + }, + json!({"type":"ws_msg","sid":"s","text":"[1]"}), + ), + ( + Out::WsClosed { + sid: id("s"), + by: Side::Gateway, + code: None, + reason: "upgrade_refused".into(), + http_status: Some(503), + }, + json!({"type":"ws_closed","sid":"s","by":"gateway","code":null,"reason":"upgrade_refused","http_status":503}), + ), + ( + Out::WsClosed { + sid: id("s"), + by: Side::Relay, + code: Some(1012), + reason: "r".into(), + http_status: None, + }, + json!({"type":"ws_closed","sid":"s","by":"relay","code":1012,"reason":"r"}), + ), + ( + Out::error(ErrorCode::DuplicateId, Some(id("q")), None), + json!({"type":"error","id":"q","code":"duplicate_id","message":"id already in use"}), + ), + ( + Out::error(ErrorCode::UnknownType, None, None), + json!({"type":"error","code":"unknown_type","message":"unknown frame type"}), + ), + ( + Out::Rotate { close_at: 5 }, + json!({"type":"rotate","close_at":5}), + ), + (Out::Ping { ts: 6.into() }, json!({"type":"ping","ts":6})), + (Out::Pong { ts: None }, json!({"type":"pong"})), + ]; + for (frame, want) in cases { + assert_eq!(serde_json::to_value(&frame).unwrap(), want); + } + } +} diff --git a/crates/buzz-relay-gateway/src/http_forward.rs b/crates/buzz-relay-gateway/src/http_forward.rs new file mode 100644 index 00000000000..72e8e037940 --- /dev/null +++ b/crates/buzz-relay-gateway/src/http_forward.rs @@ -0,0 +1,72 @@ +//! HTTP forwarding: one upstream request per `http` frame, never retried. + +use std::collections::BTreeMap; + +use axum::http::{HeaderMap, Method}; +use reqwest::{redirect::Policy, Client}; +use url::Url; + +use crate::{frame::ErrorCode, policy, Config, Limits}; + +/// The relay's answer, read up to the response cap. +pub(crate) struct Answer { + pub status: u16, + pub headers: BTreeMap<&'static str, String>, + pub body: Vec, +} + +/// The one client for the relay: no redirects, no proxy, no decompression. +pub(crate) fn client(config: &Config, limits: &Limits) -> reqwest::Result { + // The no_* calls also hold if workspace feature unification turns a codec on. + Client::builder() + .redirect(Policy::none()) + .no_proxy() + .no_gzip() + .no_brotli() + .no_deflate() + .no_zstd() + .https_only(config.relay_http.scheme() == "https") + .connect_timeout(limits.connect_timeout) + .timeout(limits.request_timeout) + .user_agent(concat!("buzz-relay-gateway/", env!("CARGO_PKG_VERSION"))) + .build() +} + +/// Sends one request and reads the whole answer, whatever its status. +pub(crate) async fn forward( + client: &Client, + method: Method, + url: Url, + headers: HeaderMap, + body: Option>, + max_body: usize, +) -> Result { + let mut request = client.request(method, url).headers(headers); + if let Some(body) = body { + request = request.body(body); + } + let mut response = request.send().await.map_err(upstream)?; + let status = response.status().as_u16(); + let headers = policy::response_headers(response.headers()); + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(upstream)? { + if body.len() + chunk.len() > max_body { + return Err(ErrorCode::ResponseTooLarge); + } + body.extend_from_slice(&chunk); + } + Ok(Answer { + status, + headers, + body, + }) +} + +fn upstream(e: reqwest::Error) -> ErrorCode { + // An expired connect_timeout is also a timeout, but then nothing reached the relay. + if e.is_timeout() && !e.is_connect() { + ErrorCode::UpstreamTimeout + } else { + ErrorCode::UpstreamConnect + } +} diff --git a/crates/buzz-relay-gateway/src/lib.rs b/crates/buzz-relay-gateway/src/lib.rs new file mode 100644 index 00000000000..1818516ca0f --- /dev/null +++ b/crates/buzz-relay-gateway/src/lib.rs @@ -0,0 +1,149 @@ +//! Buzz relay gateway: a keyless forwarder between one Worker link and one Buzz relay. +//! +//! The Worker keeps every key and signs every event; the gateway checks the +//! link bearer and forwards bytes. It holds no secret. +#![deny(unsafe_code)] + +pub mod auth; +pub mod config; +pub mod frame; +mod http_forward; +mod link; +pub mod policy; +mod relay_ws; + +use std::{ + future::IntoFuture, + sync::Arc, + time::{Duration, Instant}, +}; + +use axum::{ + extract::{ws::rejection::WebSocketUpgradeRejection, State, WebSocketUpgrade}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::get, + Json, Router, +}; +use nostr::Timestamp; +use tokio::{net::TcpListener, sync::Semaphore}; +use tokio_util::{sync::CancellationToken, task::TaskTracker}; +use tracing_subscriber::EnvFilter; + +pub use config::{Config, Limits}; + +// hyper's graceful wait never ends for a connection that stopped mid-request, +// so once the work is done the open connections get this long. +const LAST_REQUESTS: Duration = Duration::from_secs(2); + +struct AppState { + config: Config, + limits: Limits, + replay: auth::ReplayCache, + links: Arc, + shutdown: CancellationToken, + tracker: TaskTracker, + client: reqwest::Client, +} + +/// Serves `/healthz` and `/v1/link` on `listener`. Once `shutdown` is +/// cancelled both answer 503, every link drains and closes, and this returns. +pub async fn serve( + listener: TcpListener, + config: Config, + limits: Limits, + shutdown: CancellationToken, +) -> std::io::Result<()> { + let state = Arc::new(AppState { + client: http_forward::client(&config, &limits).map_err(std::io::Error::other)?, + replay: auth::ReplayCache::new(limits.replay_entries), + links: Arc::new(Semaphore::new(limits.max_links)), + shutdown: shutdown.clone(), + tracker: TaskTracker::new(), + config, + limits, + }); + let app = Router::new() + .route("/healthz", get(healthz)) + .route("/v1/link", get(link)) + .with_state(state.clone()); + // Keep accepting (and answering 503) until every link has closed. + let tracker = state.tracker.clone(); + let drained = async move { + shutdown.cancelled().await; + state.tracker.close(); + state.tracker.wait().await; + }; + let server = axum::serve(listener, app).with_graceful_shutdown(drained); + tokio::select! { + served = server.into_future() => served, + // wait() returns only after `drained` has closed the tracker and every link has ended. + () = async { + tracker.wait().await; + tokio::time::sleep(LAST_REQUESTS).await; + } => Ok(()), + } +} + +/// The log filter: `directives` in `RUST_LOG` syntax, or `info` if they do not +/// parse. The WebSocket crates stay off at any level, because at trace they +/// print whole frames: bearers, NIP-98 headers, bodies and relay messages. +pub fn log_filter(directives: &str) -> EnvFilter { + let filter = EnvFilter::try_new(directives).unwrap_or_else(|_| EnvFilter::new("info")); + // shortcut: a more specific target (tungstenite::protocol=trace) still wins; + // whoever sets RUST_LOG also picks the image, so pinning harder buys nothing. + ["tungstenite=off", "tokio_tungstenite=off"] + .into_iter() + .filter_map(|d| d.parse().ok()) + .fold(filter, EnvFilter::add_directive) +} + +async fn healthz(State(s): State>) -> Response { + if s.shutdown.is_cancelled() { + StatusCode::SERVICE_UNAVAILABLE.into_response() + } else { + "ok".into_response() + } +} + +async fn link( + State(s): State>, + headers: HeaderMap, + upgrade: Result, +) -> Response { + if s.shutdown.is_cancelled() { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + let Ok(permit) = s.links.clone().try_acquire_owned() else { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; + let claimed = auth::verify_link_bearer( + &headers, + &s.config.link_url, + &s.config.link_pubkey, + s.limits.max_auth_header, + ) + .and_then(|bearer| { + s.replay + .claim(bearer, Instant::now(), Timestamp::now().as_secs()) + }); + match (claimed, upgrade) { + (Err(reject), _) => { + tracing::warn!(reason = reject.code(), "link bearer refused"); + if reject == auth::Reject::CacheFull { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + let body = Json(serde_json::json!({ "error": "unauthorized" })); + (StatusCode::UNAUTHORIZED, body).into_response() + } + (Ok(()), Err(rejection)) => rejection.into_response(), + (Ok(()), Ok(upgrade)) => { + let max = s.limits.max_frame; + let tracked = s.tracker.clone(); + upgrade + .max_message_size(max) + .max_frame_size(max) + .on_upgrade(move |socket| tracked.track_future(link::run(socket, s, permit))) + } + } +} diff --git a/crates/buzz-relay-gateway/src/link.rs b/crates/buzz-relay-gateway/src/link.rs new file mode 100644 index 00000000000..c0036cfe7c3 --- /dev/null +++ b/crates/buzz-relay-gateway/src/link.rs @@ -0,0 +1,405 @@ +//! One task per link: `hello`, frame dispatch, pings, rotation, idle and +//! backpressure, and the teardown of everything the link opened. + +use std::{ + collections::HashMap, + sync::Arc, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use axum::extract::ws::{CloseFrame, Message, WebSocket}; +use base64::{engine::general_purpose::STANDARD, Engine as _}; +use futures_util::{stream::SplitSink, SinkExt, StreamExt}; +use tokio::{ + sync::{mpsc, mpsc::error::TrySendError, OwnedSemaphorePermit, Semaphore}, + task::{AbortHandle, JoinSet}, + time::{interval_at, sleep_until, timeout, Instant}, +}; +use tokio_util::sync::CancellationToken; + +use crate::{ + frame::{ErrorCode, HelloLimits, HttpFrame, Id, Inbound, Out}, + http_forward, policy, + relay_ws::{Cmd, Socket}, + AppState, +}; + +// Teardown waits this long for the relay sockets, then for the Worker to take the close frame. +const CLOSE_WAIT: Duration = Duration::from_secs(2); +const SHUTDOWN: (u16, &str) = (1012, "gateway restarting"); + +type Queued = (Message, Option); + +/// Milliseconds since `since`, for logs and frames. +pub(crate) fn ms(since: Instant) -> u64 { + u64::try_from(since.elapsed().as_millis()).unwrap_or(u64::MAX) +} + +fn unix_now() -> Duration { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() +} + +/// The link's bounded outbound queue. Nothing waits on it: when it is full the +/// Worker is not reading, and the link closes with 4003. +#[derive(Clone)] +pub(crate) struct Outbox { + tx: mpsc::Sender, + bytes: Arc, + overflow: CancellationToken, +} + +impl Outbox { + pub(crate) fn send(&self, frame: &Out) { + let Ok(text) = serde_json::to_string(frame) else { + return; + }; + let permit = u32::try_from(text.len()) + .ok() + .and_then(|n| self.bytes.clone().try_acquire_many_owned(n).ok()); + let full = match permit { + Some(permit) => matches!( + self.tx.try_send((Message::Text(text.into()), Some(permit))), + Err(TrySendError::Full(_)) + ), + None => true, + }; + if full { + self.overflow.cancel(); + } + } +} + +struct SocketHandle { + tx: mpsc::Sender, + overflow: CancellationToken, +} + +struct Link { + state: Arc, + id: String, + out: Outbox, + draining: bool, + http: JoinSet<()>, + inflight: HashMap, + sockets: HashMap, + socket_tasks: JoinSet<()>, + sockets_closed: CancellationToken, +} + +/// Runs one link until it closes. `_permit` holds its slot under the link cap. +pub(crate) async fn run(socket: WebSocket, state: Arc, _permit: OwnedSemaphorePermit) { + let lim = &state.limits; + let (sink, mut stream) = socket.split(); + let (tx, rx) = mpsc::channel(lim.out_frames); + let out = Outbox { + tx, + bytes: Arc::new(Semaphore::new(lim.out_bytes)), + overflow: CancellationToken::new(), + }; + let mut writer = tokio::spawn(write(sink, rx)); + let now = unix_now().as_secs(); + let rotate_in = lim.link_lifetime.saturating_sub(lim.rotate_before); + let id = uuid::Uuid::new_v4().to_string(); + out.send(&Out::Hello { + v: 1, + link_id: id.clone(), + expires_at: now + lim.link_lifetime.as_secs(), + rotate_at: now + rotate_in.as_secs(), + limits: HelloLimits { + max_request_body: lim.max_request_body, + max_response_body: lim.max_response_body, + max_ws_text: lim.max_ws_text, + max_http_inflight: lim.max_http_inflight, + max_sockets: lim.max_sockets, + }, + }); + tracing::info!(link_id = id, "link opened"); + let opened = Instant::now(); + let mut link = Link { + state: state.clone(), + id, + out, + draining: false, + http: JoinSet::new(), + inflight: HashMap::new(), + sockets: HashMap::new(), + socket_tasks: JoinSet::new(), + sockets_closed: CancellationToken::new(), + }; + let (code, reason) = link + .serve(&mut stream, opened + lim.link_lifetime, opened + rotate_in) + .await; + tracing::info!( + link_id = link.id, + code, + reason, + ms = ms(opened), + "link closing" + ); + // No relay socket outlives its link. + link.http.abort_all(); + link.sockets_closed.cancel(); + let _ = timeout(CLOSE_WAIT, async { + while link.socket_tasks.join_next().await.is_some() {} + }) + .await; + link.socket_tasks.abort_all(); + let close = Message::Close(Some(CloseFrame { + code, + reason: reason.into(), + })); + let _ = timeout(CLOSE_WAIT, link.out.tx.send((close, None))).await; + if timeout(CLOSE_WAIT, &mut writer).await.is_err() { + writer.abort(); + } +} + +async fn write(mut sink: SplitSink, mut rx: mpsc::Receiver) { + while let Some((msg, _bytes)) = rx.recv().await { + let close = matches!(msg, Message::Close(_)); + if sink.send(msg).await.is_err() || close { + break; + } + } +} + +impl Link { + async fn serve( + &mut self, + stream: &mut futures_util::stream::SplitStream, + expires: Instant, + rotate: Instant, + ) -> (u16, &'static str) { + let lim = self.state.limits.clone(); + let mut ping = interval_at(Instant::now() + lim.ping_every, lim.ping_every); + let mut idle = Instant::now() + lim.idle; + let mut rotated = false; + let mut drain_until = None; + loop { + tokio::select! { + msg = stream.next() => { + idle = Instant::now() + lim.idle; + match msg { + Some(Ok(Message::Text(text))) => { + if let Err(close) = self.on_text(text.as_str()) { + return close; + } + } + Some(Ok(Message::Binary(_))) => return (1003, "binary frames are reserved"), + Some(Ok(Message::Close(_))) | None => return (1000, "closed by worker"), + Some(Ok(_)) => {} + // axum's tungstenite is not ours to downcast; its capacity error reads this way. + Some(Err(e)) if e.to_string().contains("Space limit exceeded") => return (1009, "frame too large"), + Some(Err(_)) => return (1002, "protocol error"), + } + } + _ = ping.tick() => { + let ts = u64::try_from(unix_now().as_millis()).unwrap_or(u64::MAX); + self.out.send(&Out::Ping { ts: ts.into() }); + } + _ = sleep_until(rotate), if !rotated => { + rotated = true; + let close_at = (unix_now() + expires.saturating_duration_since(Instant::now())).as_secs(); + self.out.send(&Out::Rotate { close_at }); + } + _ = sleep_until(expires) => return (4001, "rotate"), + _ = sleep_until(idle) => return (4002, "idle"), + _ = self.out.overflow.cancelled() => return (4003, "backpressure"), + _ = self.state.shutdown.cancelled(), if !self.draining => { + self.draining = true; + drain_until = Some(Instant::now() + lim.shutdown_grace); + if self.http.is_empty() { + return SHUTDOWN; + } + } + _ = sleep_until(drain_until.unwrap_or(expires)), if drain_until.is_some() => return SHUTDOWN, + Some(_) = self.http.join_next(), if !self.http.is_empty() => { + self.inflight.retain(|_, task| !task.is_finished()); + if self.draining && self.http.is_empty() { + return SHUTDOWN; + } + } + Some(_) = self.socket_tasks.join_next(), if !self.socket_tasks.is_empty() => { + self.sockets.retain(|_, socket| !socket.tx.is_closed()); + } + } + } + } + + /// Handles one Worker frame. `Err` closes the link with that code. + fn on_text(&mut self, text: &str) -> Result<(), (u16, &'static str)> { + let Ok(frame) = Inbound::parse(text) else { + self.refuse(ErrorCode::BadFrame, None, None); + return Err((1008, "bad frame")); + }; + match frame { + Inbound::Http(f) => { + let id = f.id.clone(); + if let Err(code) = self.http(f) { + self.refuse(code, Some(id), None); + } + } + Inbound::WsOpen { sid, path } => { + if let Err(code) = self.ws_open(sid.clone(), &path) { + self.refuse(code, None, Some(sid)); + } + } + Inbound::WsSend { sid, text } => { + let result = if text.len() > self.state.limits.max_ws_text { + Err(ErrorCode::FrameTooLarge) + } else { + self.to_socket(&sid, Cmd::Send(text)) + }; + if let Err(code) = result { + self.refuse(code, None, Some(sid)); + } + } + Inbound::WsClose { sid, code, reason } => { + let code = code.unwrap_or(1000); + let reason = reason.unwrap_or_default(); + // A close frame allows 1000 or an application code, and a 123-byte reason. + let result = + if (code == 1000 || (3000..=4999).contains(&code)) && reason.len() <= 123 { + self.to_socket(&sid, Cmd::Close(code, reason)) + } else { + Err(ErrorCode::NotAllowed) + }; + if let Err(code) = result { + self.refuse(code, None, Some(sid)); + } + } + Inbound::Ping { ts } => self.out.send(&Out::Pong { ts }), + Inbound::Pong => {} + Inbound::Unknown => self.refuse(ErrorCode::UnknownType, None, None), + } + Ok(()) + } + + fn refuse(&self, code: ErrorCode, id: Option, sid: Option) { + tracing::info!( + link_id = self.id, + id = id.as_ref().map(Id::as_str), + sid = sid.as_ref().map(Id::as_str), + code = ?code, + "frame refused" + ); + self.out.send(&Out::error(code, id, sid)); + } + + fn http(&mut self, f: HttpFrame) -> Result<(), ErrorCode> { + let lim = &self.state.limits; + if self.draining { + return Err(ErrorCode::ShuttingDown); + } + if self.inflight.contains_key(&f.id) { + return Err(ErrorCode::DuplicateId); + } + if self.inflight.len() >= lim.max_http_inflight { + return Err(ErrorCode::Busy); + } + let route = policy::http_route(&f.method, &f.path)?; + let raw_headers = f.headers.as_deref().map(|raw| raw.get()); + let headers = policy::request_headers(route, raw_headers, lim.max_auth_header)?; + let body = match (route.method() == axum::http::Method::POST, f.body_b64) { + (true, Some(b64)) => { + let body = STANDARD.decode(b64).map_err(|_| ErrorCode::BadBase64)?; + if body.len() > lim.max_request_body { + return Err(ErrorCode::BodyTooLarge); + } + Some(body) + } + (false, None) => None, + _ => return Err(ErrorCode::NotAllowed), + }; + let url = self + .state + .config + .relay_http + .join(route.path()) + .map_err(|_| ErrorCode::NotAllowed)?; + let (state, out, link_id, id) = ( + self.state.clone(), + self.out.clone(), + self.id.clone(), + f.id.clone(), + ); + let task = self.http.spawn(async move { + let started = Instant::now(); + let result = http_forward::forward( + &state.client, + route.method(), + url, + headers, + body, + state.limits.max_response_body, + ) + .await; + let (link_id, path, ms) = (link_id.as_str(), route.path(), ms(started)); + out.send(&match result { + Ok(answer) => { + tracing::info!(link_id, id = id.as_str(), path, status = answer.status, ms, "http forwarded"); + let body_b64 = STANDARD.encode(answer.body); + Out::HttpRes { id, status: answer.status, headers: answer.headers, body_b64, ms } + } + Err(code) => { + tracing::info!(link_id, id = id.as_str(), path, code = ?code, ms, "http failed"); + Out::error(code, Some(id), None) + } + }); + }); + self.inflight.insert(f.id, task); + Ok(()) + } + + fn ws_open(&mut self, sid: Id, path: &str) -> Result<(), ErrorCode> { + let lim = &self.state.limits; + if self.draining { + return Err(ErrorCode::ShuttingDown); + } + if self.sockets.contains_key(&sid) { + return Err(ErrorCode::DuplicateId); + } + if self.sockets.len() >= lim.max_sockets { + return Err(ErrorCode::Busy); + } + let url = self + .state + .config + .relay_ws + .join(policy::ws_route(path)?) + .map_err(|_| ErrorCode::NotAllowed)?; + let (tx, cmds) = mpsc::channel(lim.relay_queue); + let overflow = CancellationToken::new(); + let socket = Socket { + link_id: self.id.clone(), + sid: sid.clone(), + url, + out: self.out.clone(), + cmds, + overflow: overflow.clone(), + link_closed: self.sockets_closed.clone(), + shutdown: self.state.shutdown.clone(), + connect_timeout: lim.connect_timeout, + silence: lim.relay_silence, + max_frame: lim.max_relay_frame, + }; + self.socket_tasks.spawn(socket.run()); + self.sockets.insert(sid, SocketHandle { tx, overflow }); + Ok(()) + } + + fn to_socket(&self, sid: &Id, cmd: Cmd) -> Result<(), ErrorCode> { + let socket = self.sockets.get(sid).ok_or(ErrorCode::UnknownSocket)?; + match socket.tx.try_send(cmd) { + Ok(()) => Ok(()), + Err(TrySendError::Closed(_)) => Err(ErrorCode::UnknownSocket), + // The relay is not taking frames: close that socket, not the link. + Err(TrySendError::Full(_)) => { + socket.overflow.cancel(); + Ok(()) + } + } + } +} diff --git a/crates/buzz-relay-gateway/src/main.rs b/crates/buzz-relay-gateway/src/main.rs new file mode 100644 index 00000000000..792d4206d03 --- /dev/null +++ b/crates/buzz-relay-gateway/src/main.rs @@ -0,0 +1,56 @@ +use buzz_relay_gateway::{Config, Limits}; +use tokio::net::TcpListener; +use tokio_util::sync::CancellationToken; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Workspace builds unify ring and aws-lc-rs; rustls then needs an explicit provider. + let _ = rustls::crypto::ring::default_provider().install_default(); + let directives = std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_owned()); + tracing_subscriber::fmt() + .json() + .with_env_filter(buzz_relay_gateway::log_filter(&directives)) + .init(); + let config = match Config::from_env() { + Ok(config) => config, + Err(e) => { + tracing::error!(error = %e, "invalid configuration"); + std::process::exit(1); + } + }; + tracing::info!( + bind = %config.bind_addr, + relay = config.relay_ws.host_str().unwrap_or_default(), + link_url = %config.link_url, + link_pubkey = %config.link_pubkey, + "relay gateway starting" + ); + let listener = TcpListener::bind(config.bind_addr).await?; + let shutdown = CancellationToken::new(); + tokio::spawn({ + let shutdown = shutdown.clone(); + async move { + signal().await; + tracing::info!("shutdown signal received"); + shutdown.cancel(); + } + }); + buzz_relay_gateway::serve(listener, config, Limits::default(), shutdown).await?; + Ok(()) +} + +// The binary is PID 1 in the image, which has no default signal handling. +async fn signal() { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + if let Ok(mut term) = signal(SignalKind::terminate()) { + tokio::select! { + _ = term.recv() => {} + _ = tokio::signal::ctrl_c() => {} + } + return; + } + } + let _ = tokio::signal::ctrl_c().await; +} diff --git a/crates/buzz-relay-gateway/src/policy.rs b/crates/buzz-relay-gateway/src/policy.rs new file mode 100644 index 00000000000..3cfb75c2b25 --- /dev/null +++ b/crates/buzz-relay-gateway/src/policy.rs @@ -0,0 +1,249 @@ +//! What may reach the relay: an exact route table and closed header sets (design §4). + +use std::collections::BTreeMap; + +use axum::http::{header, HeaderMap, HeaderValue, Method}; +use serde::Deserialize; + +const NIP11: &str = "application/nostr+json"; + +/// A relay HTTP route. The upstream path is always one of these literals. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum HttpRoute { + /// `POST /query` + Query, + /// `POST /events` + Events, + /// `POST /count` + Count, + /// `GET /` for the NIP-11 relay information document. + Info, +} + +impl HttpRoute { + /// The exact path sent upstream. + pub fn path(self) -> &'static str { + match self { + Self::Query => "/query", + Self::Events => "/events", + Self::Count => "/count", + Self::Info => "/", + } + } + + /// The upstream method; every POST route carries a body and GET carries none. + pub fn method(self) -> Method { + if self == Self::Info { + Method::GET + } else { + Method::POST + } + } +} + +/// Why a frame may not go upstream. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Refused { + /// Not in the table, or a header that is not allowed. + NotAllowed, + /// A route reserved for a later phase (the huddle audio path). + NotSupported, +} + +/// Matches a frame's method and path exactly: no normalisation, no query string. +pub fn http_route(method: &str, path: &str) -> Result { + match (method, path) { + ("POST", "/query") => Ok(HttpRoute::Query), + ("POST", "/events") => Ok(HttpRoute::Events), + ("POST", "/count") => Ok(HttpRoute::Count), + ("GET", "/") => Ok(HttpRoute::Info), + _ => Err(Refused::NotAllowed), + } +} + +/// Relay socket paths: `/` only. The huddle audio path is reserved for Phase 2. +pub fn ws_route(path: &str) -> Result<&'static str, Refused> { + if path == "/" { + return Ok("/"); + } + let audio = path + .strip_prefix("/huddle/") + .and_then(|rest| rest.strip_suffix("/audio")) + .is_some_and(|id| id.len() == 36 && uuid::Uuid::try_parse(id).is_ok()); + Err(if audio { + Refused::NotSupported + } else { + Refused::NotAllowed + }) +} + +// A struct, not a map: serde refuses unknown and duplicated keys for it. +#[derive(Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct FrameHeaders { + authorization: Option, + #[serde(rename = "content-type")] + content_type: Option, + accept: Option, +} + +/// Builds the upstream request headers from a frame's raw `headers` object. +/// POST routes need `authorization`; `GET /` needs exactly the NIP-11 `accept`. +pub fn request_headers( + route: HttpRoute, + raw: Option<&str>, + max_auth: usize, +) -> Result { + let h: FrameHeaders = match raw { + Some(raw) => serde_json::from_str(raw).map_err(|_| Refused::NotAllowed)?, + None => FrameHeaders::default(), + }; + let ok = match route { + HttpRoute::Info => { + h.authorization.is_none() + && h.content_type.is_none() + && h.accept.as_deref() == Some(NIP11) + } + _ => h + .authorization + .as_ref() + .is_some_and(|a| a.len() <= max_auth), + }; + if !ok { + return Err(Refused::NotAllowed); + } + let mut out = HeaderMap::new(); + for (name, value) in [ + (header::AUTHORIZATION, h.authorization), + (header::CONTENT_TYPE, h.content_type), + (header::ACCEPT, h.accept), + ] { + if let Some(value) = value { + out.insert( + name, + HeaderValue::from_str(&value).map_err(|_| Refused::NotAllowed)?, + ); + } + } + Ok(out) +} + +/// The relay response headers the Worker sees: `content-type` and `retry-after` only. +pub fn response_headers(headers: &HeaderMap) -> BTreeMap<&'static str, String> { + [ + ("content-type", header::CONTENT_TYPE), + ("retry-after", header::RETRY_AFTER), + ] + .into_iter() + .filter_map(|(key, name)| Some((key, headers.get(name)?.to_str().ok()?.to_owned()))) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn route_table_is_exact() { + assert_eq!(http_route("POST", "/query"), Ok(HttpRoute::Query)); + assert_eq!(http_route("POST", "/events"), Ok(HttpRoute::Events)); + assert_eq!(http_route("POST", "/count"), Ok(HttpRoute::Count)); + assert_eq!(http_route("GET", "/"), Ok(HttpRoute::Info)); + for (m, p) in [ + ("GET", "/query"), + ("PUT", "/query"), + ("post", "/query"), + ("POST", "/QUERY"), + ("POST", "/query/"), + ("POST", "//query"), + ("POST", "/query?x=1"), + ("POST", "/./query"), + ("POST", "/%71uery"), + ("POST", "/"), + ("GET", "/hooks/1"), + ] { + assert_eq!(http_route(m, p), Err(Refused::NotAllowed), "{m} {p}"); + } + assert_eq!(ws_route("/"), Ok("/")); + assert_eq!( + ws_route("/huddle/6f1c2c1e-4b9a-4c71-9a55-0f1d8c3a2b10/audio"), + Err(Refused::NotSupported) + ); + for p in [ + "", + "/query", + "//", + "/?x", + "/huddle/x/audio", + "/huddle/6f1c2c1e4b9a4c719a550f1d8c3a2b10/audio", + ] { + assert_eq!(ws_route(p), Err(Refused::NotAllowed), "{p}"); + } + } + + #[test] + fn request_headers_are_a_closed_set() { + let post = |raw: &str| request_headers(HttpRoute::Query, Some(raw), 16); + let h = post(r#"{"authorization":"Nostr abc","content-type":"application/json"}"#).unwrap(); + assert_eq!(h.len(), 2); + assert_eq!(h[header::AUTHORIZATION], "Nostr abc"); + for bad in [ + r#"{"authorization":"Nostr abc","x-pubkey":"00"}"#, + r#"{"authorization":"Nostr abc","host":"evil.example"}"#, + r#"{"authorization":"Nostr abc","x-forwarded-for":"1.2.3.4"}"#, + r#"{"authorization":"Nostr abc","accept":"a","accept":"b"}"#, + r#"{"authorization":"Nostr abc\r\nx-pubkey: 00"}"#, + r#"{"authorization":"Nostr abc","content-type":"a\nb"}"#, + r#"{"Authorization":"Nostr abc"}"#, + r#"{"authorization":"Nostr 0123456789abc"}"#, + r#"{"content-type":"application/json"}"#, + r#"["authorization"]"#, + ] { + assert_eq!(post(bad), Err(Refused::NotAllowed), "{bad}"); + } + assert_eq!( + request_headers(HttpRoute::Query, None, 16), + Err(Refused::NotAllowed) + ); + let info = |raw: &str| request_headers(HttpRoute::Info, Some(raw), 16); + assert!(info(r#"{"accept":"application/nostr+json"}"#).is_ok()); + assert_eq!( + info(r#"{"accept":"application/json"}"#), + Err(Refused::NotAllowed) + ); + assert_eq!( + info(r#"{"accept":"application/nostr+json","authorization":"x"}"#), + Err(Refused::NotAllowed) + ); + assert_eq!( + request_headers(HttpRoute::Info, None, 16), + Err(Refused::NotAllowed) + ); + } + + #[test] + fn response_headers_keep_two_names() { + let mut h = HeaderMap::new(); + for (name, value) in [ + ("content-type", "application/json"), + ("retry-after", "12"), + ("set-cookie", "a=b"), + ("location", "/elsewhere"), + ("server", "cloudflare"), + ("cf-ray", "1"), + ] { + h.insert( + header::HeaderName::from_static(name), + HeaderValue::from_static(value), + ); + } + let kept = response_headers(&h); + assert_eq!( + kept.into_iter().collect::>(), + [ + ("content-type", "application/json".into()), + ("retry-after", "12".into()) + ] + ); + } +} diff --git a/crates/buzz-relay-gateway/src/relay_ws.rs b/crates/buzz-relay-gateway/src/relay_ws.rs new file mode 100644 index 00000000000..641a9d091de --- /dev/null +++ b/crates/buzz-relay-gateway/src/relay_ws.rs @@ -0,0 +1,207 @@ +//! One task per relay socket. It keeps reading the relay, so the relay's pings +//! are answered at once, and it reports exactly one `ws_closed`. + +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use tokio::{ + sync::mpsc, + time::{sleep_until, timeout, Instant}, +}; +use tokio_tungstenite::{ + connect_async_with_config, + tungstenite::{ + client::IntoClientRequest, + http::{header::USER_AGENT, HeaderValue}, + protocol::{CloseFrame, WebSocketConfig}, + Error, Message, + }, +}; +use tokio_util::sync::CancellationToken; +use url::Url; + +use crate::{ + frame::{Id, Out, Side}, + link::{ms, Outbox}, +}; + +const AUTH_PREFIX: &str = r#"["AUTH","#; +const CLOSE_CAP: Duration = Duration::from_secs(1); + +/// A Worker frame for one relay socket. +pub(crate) enum Cmd { + Send(String), + Close(u16, String), +} + +/// One relay socket. The link keeps the sending half of `cmds`. +pub(crate) struct Socket { + pub link_id: String, + pub sid: Id, + pub url: Url, + pub out: Outbox, + pub cmds: mpsc::Receiver, + /// Cancelled when the link could not queue a Worker frame for this socket. + pub overflow: CancellationToken, + /// Cancelled when the link ends. + pub link_closed: CancellationToken, + pub shutdown: CancellationToken, + pub connect_timeout: Duration, + pub silence: Duration, + pub max_frame: usize, +} + +struct Closed { + by: Side, + code: Option, + reason: String, + http_status: Option, +} + +fn by_gateway(code: Option, reason: &str) -> Closed { + Closed { + by: Side::Gateway, + code, + reason: reason.to_owned(), + http_status: None, + } +} + +impl Socket { + pub(crate) async fn run(mut self) { + let opened = Instant::now(); + let closed = self.serve(opened).await; + // Relay and Worker close reasons are frame content, so only gateway reasons are logged. + let reason = (closed.by == Side::Gateway).then_some(closed.reason.as_str()); + tracing::info!( + link_id = self.link_id, + sid = self.sid.as_str(), + by = ?closed.by, + code = closed.code, + reason, + http_status = closed.http_status, + ms = ms(opened), + "relay socket closed" + ); + self.out.send(&Out::WsClosed { + sid: self.sid, + by: closed.by, + code: closed.code, + reason: closed.reason, + http_status: closed.http_status, + }); + } + + fn end(&self) -> (u16, &'static str) { + if self.shutdown.is_cancelled() { + (1001, "shutdown") + } else { + (1000, "link_closed") + } + } + + async fn serve(&mut self, opened: Instant) -> Closed { + let Ok(mut request) = self.url.as_str().into_client_request() else { + return by_gateway(None, "connect_failed"); + }; + let agent = concat!("buzz-relay-gateway/", env!("CARGO_PKG_VERSION")); + request + .headers_mut() + .insert(USER_AGENT, HeaderValue::from_static(agent)); + let config = WebSocketConfig::default() + .max_message_size(Some(self.max_frame)) + .max_frame_size(Some(self.max_frame)); + let connecting = connect_async_with_config(request, Some(config), false); + let connect = timeout(self.connect_timeout, connecting); + let mut ws = tokio::select! { + result = connect => match result { + Err(_) => return by_gateway(None, "connect_timeout"), + Ok(Err(Error::Http(response))) => { + let status = Some(response.status().as_u16()); + return Closed { http_status: status, ..by_gateway(None, "upgrade_refused") }; + } + Ok(Err(_)) => return by_gateway(None, "connect_failed"), + Ok(Ok((ws, _))) => ws, + }, + _ = self.link_closed.cancelled() => return by_gateway(None, self.end().1), + }; + self.out.send(&Out::WsOpened { + sid: self.sid.clone(), + ms: ms(opened), + }); + let (link_id, sid) = (self.link_id.as_str(), self.sid.as_str()); + let mut challenge = None; + let mut silent_at = Instant::now() + self.silence; + let (frame, closed) = loop { + let (code, reason) = tokio::select! { + msg = ws.next() => { + silent_at = Instant::now() + self.silence; + match msg { + Some(Ok(Message::Text(text))) => { + if text.starts_with(AUTH_PREFIX) { + tracing::info!(link_id, sid, auth_challenge_ms = ms(opened), "relay sent AUTH"); + challenge = Some(Instant::now()); + } + self.out.send(&Out::WsMsg { sid: self.sid.clone(), text: text.as_str().to_owned() }); + continue; + } + Some(Ok(Message::Close(frame))) => { + let (code, reason) = frame.map_or((None, String::new()), |f| (Some(f.code.into()), f.reason.as_str().to_owned())); + // tungstenite only queues the close reply; flushing sends it. + let _ = timeout(CLOSE_CAP, ws.flush()).await; + return Closed { by: Side::Relay, code, reason, http_status: None }; + } + Some(Ok(_)) => continue, + Some(Err(Error::Capacity(_))) => (1009, "frame_too_large"), + Some(Err(_)) | None => { + return Closed { by: Side::Relay, code: None, reason: "connection_lost".into(), http_status: None }; + } + } + } + cmd = self.cmds.recv() => match cmd { + Some(Cmd::Send(text)) => { + if let Some(at) = challenge.filter(|_| text.starts_with(AUTH_PREFIX)) { + tracing::info!(link_id, sid, auth_reply_ms = ms(at), "worker answered AUTH"); + challenge = None; + } + // A relay that stops reading stalls this write, so the socket's other ends still apply. + tokio::select! { + sent = ws.send(Message::Text(text.into())) => { + if sent.is_err() { + return Closed { by: Side::Relay, code: None, reason: "connection_lost".into(), http_status: None }; + } + continue; + } + _ = self.overflow.cancelled() => (1000, "backpressure"), + _ = self.link_closed.cancelled() => self.end(), + _ = sleep_until(silent_at) => (1001, "relay_silent"), + } + } + Some(Cmd::Close(code, reason)) => { + let frame = CloseFrame { code: code.into(), reason: reason.clone().into() }; + break (frame, Closed { by: Side::Worker, code: Some(code), reason, http_status: None }); + } + None => self.end(), + }, + _ = self.overflow.cancelled() => (1000, "backpressure"), + _ = self.link_closed.cancelled() => self.end(), + _ = sleep_until(silent_at) => (1001, "relay_silent"), + }; + break ( + CloseFrame { + code: code.into(), + reason: reason.into(), + }, + by_gateway(Some(code), reason), + ); + }; + // The relay's close reply ends the stream; capped so teardown stays bounded. + let _ = timeout(CLOSE_CAP, async { + if ws.close(Some(frame)).await.is_ok() { + while let Some(Ok(_)) = ws.next().await {} + } + }) + .await; + closed + } +} diff --git a/crates/buzz-relay-gateway/tests/common/mod.rs b/crates/buzz-relay-gateway/tests/common/mod.rs new file mode 100644 index 00000000000..4397aa0d491 --- /dev/null +++ b/crates/buzz-relay-gateway/tests/common/mod.rs @@ -0,0 +1,446 @@ +//! Shared harness: the gateway in-process, a mock relay, a mock Worker and a bearer signer. +#![allow(dead_code)] + +use std::{ + collections::HashMap, + net::SocketAddr, + sync::{Arc, Mutex}, + time::Duration, +}; + +use axum::{ + body::{Body, Bytes}, + extract::{ + ws::{rejection::WebSocketUpgradeRejection, Message as AxMessage, WebSocket}, + State, WebSocketUpgrade, + }, + http::{HeaderMap, Method, StatusCode, Uri}, + response::{IntoResponse, Response}, + Router, +}; +use base64::{engine::general_purpose::STANDARD, Engine as _}; +use buzz_relay_gateway::{Config, Limits}; +use futures_util::{SinkExt, StreamExt}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use tokio::{net::TcpListener, task::JoinHandle, time::timeout}; +use tokio_tungstenite::{ + connect_async, + tungstenite::{client::IntoClientRequest, Message}, + MaybeTlsStream, WebSocketStream, +}; +use tokio_util::sync::CancellationToken; + +pub const LINK_URL: &str = "http://gateway.test/v1/link"; + +/// A fresh link bearer signed by `keys`, carrying `content`. +pub fn bearer_with(keys: &Keys, content: &str) -> String { + let nonce = uuid::Uuid::new_v4().to_string(); + let tags = [ + ["u", LINK_URL], + ["method", "GET"], + ["nonce", nonce.as_str()], + ] + .map(|t| Tag::parse(t).unwrap()); + let event = EventBuilder::new(Kind::HttpAuth, content) + .tags(tags) + .sign_with_keys(keys) + .unwrap(); + format!( + "Nostr {}", + STANDARD.encode(serde_json::to_string(&event).unwrap()) + ) +} + +pub fn bearer(keys: &Keys) -> String { + bearer_with(keys, "") +} + +/// Short timers for tests; every other limit stays at its default. +pub fn limits() -> Limits { + Limits { + request_timeout: Duration::from_secs(2), + shutdown_grace: Duration::from_millis(500), + ..Limits::default() + } +} + +pub struct Gateway { + pub addr: SocketAddr, + pub keys: Keys, + pub shutdown: CancellationToken, + pub task: JoinHandle>, +} + +impl Gateway { + pub async fn start(relay_ws: &str, limits: Limits) -> Self { + let keys = Keys::generate(); + let pubkey = keys.public_key().to_hex(); + let config = Config::from_lookup(|name| match name { + "BUZZ_RELAY_GATEWAY_RELAY_URL" => Some(relay_ws.to_owned()), + "BUZZ_RELAY_GATEWAY_LINK_URL" => Some(LINK_URL.to_owned()), + "BUZZ_RELAY_GATEWAY_LINK_PUBKEY" => Some(pubkey.clone()), + _ => None, + }) + .unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let shutdown = CancellationToken::new(); + let task = tokio::spawn(buzz_relay_gateway::serve( + listener, + config, + limits, + shutdown.clone(), + )); + Self { + addr, + keys, + shutdown, + task, + } + } + + pub fn url(&self, path: &str) -> String { + format!("http://{}{path}", self.addr) + } + + /// A Worker on a fresh link, past `hello`. + pub async fn worker(&self) -> Worker { + Worker::connect(self, &bearer(&self.keys)).await.unwrap() + } +} + +/// The mock Worker: one link, frames as JSON values. +pub struct Worker { + pub ws: WebSocketStream>, + pub hello: Value, +} + +pub enum Next { + Frame(Value), + Closed(Option), +} + +impl Worker { + pub async fn connect( + gw: &Gateway, + bearer: &str, + ) -> Result { + let mut request = format!("ws://{}/v1/link", gw.addr).into_client_request()?; + request + .headers_mut() + .insert("authorization", bearer.parse().unwrap()); + let (ws, _) = connect_async(request).await?; + let mut worker = Self { + ws, + hello: Value::Null, + }; + worker.hello = worker.frame().await; + assert_eq!(worker.hello["type"], "hello"); + Ok(worker) + } + + pub async fn send(&mut self, frame: Value) { + self.ws + .send(Message::text(frame.to_string())) + .await + .unwrap(); + } + + /// The next frame other than a gateway `ping`. + pub async fn next(&mut self) -> Next { + loop { + let msg = timeout(Duration::from_secs(10), self.ws.next()) + .await + .expect("no frame within 10 s"); + match msg { + Some(Ok(Message::Text(text))) => { + let frame: Value = serde_json::from_str(&text).unwrap(); + if frame["type"] != "ping" { + return Next::Frame(frame); + } + } + Some(Ok(Message::Close(frame))) => { + return Next::Closed(frame.map(|f| f.code.into())) + } + Some(Ok(_)) => {} + Some(Err(_)) | None => return Next::Closed(None), + } + } + } + + pub async fn frame(&mut self) -> Value { + match self.next().await { + Next::Frame(frame) => frame, + Next::Closed(code) => panic!("link closed: {code:?}"), + } + } + + /// Reads until the link closes; returns the close code. + pub async fn closed(&mut self) -> Option { + loop { + if let Next::Closed(code) = self.next().await { + return code; + } + } + } + + pub async fn http( + &mut self, + id: &str, + method: &str, + path: &str, + headers: Value, + body: Option<&[u8]>, + ) -> Value { + let mut frame = + json!({ "type": "http", "id": id, "method": method, "path": path, "headers": headers }); + if let Some(body) = body { + frame["body_b64"] = STANDARD.encode(body).into(); + } + self.send(frame).await; + self.frame().await + } + + pub async fn post(&mut self, id: &str, path: &str, body: &[u8]) -> Value { + self.http( + id, + "POST", + path, + json!({ "authorization": "Nostr test" }), + Some(body), + ) + .await + } + + /// Opens relay socket `sid` and returns the relay's AUTH challenge. + pub async fn open(&mut self, sid: &str) -> String { + self.send(json!({"type": "ws_open", "sid": sid, "path": "/"})) + .await; + let opened = self.frame().await; + assert_eq!( + (opened["type"].as_str(), opened["sid"].as_str()), + (Some("ws_opened"), Some(sid)) + ); + let msg = self.frame().await; + assert_eq!( + (msg["type"].as_str(), msg["sid"].as_str()), + (Some("ws_msg"), Some(sid)) + ); + let auth: Value = serde_json::from_str(msg["text"].as_str().unwrap()).unwrap(); + assert_eq!(auth[0], "AUTH"); + auth[1].as_str().unwrap().to_owned() + } + + /// Sends `text` on relay socket `sid` without waiting for an answer. + pub async fn ws_send(&mut self, sid: &str, text: &str) { + self.send(json!({"type": "ws_send", "sid": sid, "text": text})) + .await; + } + + /// Sends `text` on socket `sid` and returns the relay's next frame on the link. + pub async fn say(&mut self, sid: &str, text: Value) -> Value { + self.ws_send(sid, &text.to_string()).await; + let msg = self.frame().await; + assert_eq!(msg["type"], "ws_msg", "{msg}"); + serde_json::from_str(msg["text"].as_str().unwrap()).unwrap() + } +} + +/// A mock relay, a gateway pointed at it, and a Worker on a fresh link. +pub async fn setup(limits: Limits) -> (Relay, Gateway, Worker) { + let relay = Relay::start().await; + let gw = Gateway::start(&relay.ws_url(), limits).await; + let w = gw.worker().await; + (relay, gw, w) +} + +pub fn body_of(frame: &Value) -> Vec { + STANDARD + .decode(frame["body_b64"].as_str().unwrap()) + .unwrap() +} + +pub fn sha256_hex(bytes: &[u8]) -> String { + hex::encode(Sha256::digest(bytes)) +} + +/// How the mock relay answers HTTP (and socket upgrades, for `Reply`). +#[derive(Clone, Default)] +pub enum Mode { + /// Report the hash, length and headers of what arrived. + #[default] + Echo, + /// This status and body. + Reply(u16, Vec), + /// 302 to `/elsewhere`. + Redirect, + /// Never answer. + Stall, + /// Send part of a body, then drop the connection. + CloseMid, +} + +#[derive(Default)] +pub struct RelayState { + pub mode: Mutex, + pub hits: Mutex>, + pub open_sockets: Mutex, + pub socket_closes: Mutex>>, + pub url: Mutex, +} + +/// The mock relay: HTTP routes on every path and NIP-42 sockets on `/`. +pub struct Relay { + pub addr: SocketAddr, + pub state: Arc, +} + +impl Relay { + pub async fn start() -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let state = Arc::new(RelayState::default()); + *state.url.lock().unwrap() = format!("ws://{addr}"); + let app = Router::new().fallback(handle).with_state(state.clone()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + Self { addr, state } + } + + pub fn ws_url(&self) -> String { + format!("ws://{}", self.addr) + } + + pub fn set(&self, mode: Mode) { + *self.state.mode.lock().unwrap() = mode; + } + + pub fn hits(&self, path: &str) -> usize { + self.state + .hits + .lock() + .unwrap() + .get(path) + .copied() + .unwrap_or(0) + } + + pub fn open_sockets(&self) -> usize { + *self.state.open_sockets.lock().unwrap() + } + + pub fn socket_closes(&self) -> Vec> { + self.state.socket_closes.lock().unwrap().clone() + } +} + +async fn handle( + State(s): State>, + method: Method, + uri: Uri, + headers: HeaderMap, + upgrade: Result, + body: Bytes, +) -> Response { + let path = uri.path().to_owned(); + *s.hits.lock().unwrap().entry(path.clone()).or_default() += 1; + let mode = s.mode.lock().unwrap().clone(); + if let (Ok(upgrade), Mode::Echo) = (upgrade, &mode) { + return upgrade.on_upgrade(move |ws| relay_socket(ws, s)); + } + match mode { + Mode::Echo if method == Method::GET => { + json!({ "name": "mock relay", "accept": headers["accept"].to_str().unwrap() }) + .to_string() + .into_response() + } + Mode::Echo => { + let seen: serde_json::Map = headers + .iter() + .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").into())) + .collect(); + let url = format!( + "http://{}{path}", + s.url.lock().unwrap().trim_start_matches("ws://") + ); + let nip98 = headers + .get("authorization") + .and_then(|v| v.to_str().ok()?.strip_prefix("Nostr ").map(str::to_owned)) + .and_then(|b64| String::from_utf8(STANDARD.decode(b64).ok()?).ok()) + .is_some_and(|event| { + buzz_auth::verify_nip98_event(&event, &url, "POST", Some(&body)).is_ok() + }); + let echo = json!({ "sha256": sha256_hex(&body), "len": body.len(), "headers": seen, "nip98": nip98 }); + ([("content-type", "application/json")], echo.to_string()).into_response() + } + Mode::Reply(status, body) => ( + StatusCode::from_u16(status).unwrap(), + [("set-cookie", "a=b")], + body, + ) + .into_response(), + Mode::Redirect => (StatusCode::FOUND, [("location", "/elsewhere")]).into_response(), + Mode::Stall => { + tokio::time::sleep(Duration::from_secs(30)).await; + StatusCode::OK.into_response() + } + Mode::CloseMid => { + let chunks = [ + Ok(Bytes::from_static(b"partial")), + Err(std::io::Error::other("cut")), + ]; + Body::from_stream(futures_util::stream::iter(chunks)).into_response() + } + } +} + +/// A relay socket: AUTH challenge first, then commands from the test. +async fn relay_socket(mut ws: WebSocket, s: Arc) { + *s.open_sockets.lock().unwrap() += 1; + let challenge = buzz_auth::generate_challenge(); + let relay_url = s.url.lock().unwrap().clone(); + let reply = |v: Value| AxMessage::text(v.to_string()); + let _ = ws.send(reply(json!(["AUTH", challenge]))).await; + while let Some(Ok(msg)) = ws.recv().await { + let AxMessage::Text(text) = msg else { + if let AxMessage::Close(frame) = msg { + s.socket_closes.lock().unwrap().push(frame.map(|f| f.code)); + break; + } + continue; + }; + let cmd: Value = serde_json::from_str(&text).unwrap_or_default(); + let out = match cmd[0].as_str() { + Some("AUTH") => { + let event: nostr::Event = serde_json::from_value(cmd[1].clone()).unwrap(); + let ok = buzz_auth::verify_nip42_event(&event, &challenge, &relay_url).is_ok(); + vec![reply(json!(["OK", event.id, ok, ""]))] + } + Some("REQ") => vec![reply(json!(["EOSE", cmd[1]]))], + Some("CLOSE_WITH") => { + let code = cmd[1].as_u64().unwrap() as u16; + let frame = axum::extract::ws::CloseFrame { + code, + reason: "relay restarting".into(), + }; + vec![AxMessage::Close(Some(frame))] + } + Some("SEND") => { + let (n, size) = (cmd[1].as_u64().unwrap(), cmd[2].as_u64().unwrap() as usize); + (0..n).map(|_| AxMessage::text("x".repeat(size))).collect() + } + Some("SILENT") => { + tokio::time::sleep(Duration::from_secs(30)).await; + vec![] + } + _ => vec![], + }; + for msg in out { + if ws.send(msg).await.is_err() { + break; + } + } + } + *s.open_sockets.lock().unwrap() -= 1; +} diff --git a/crates/buzz-relay-gateway/tests/http.rs b/crates/buzz-relay-gateway/tests/http.rs new file mode 100644 index 00000000000..219a0486e63 --- /dev/null +++ b/crates/buzz-relay-gateway/tests/http.rs @@ -0,0 +1,246 @@ +mod common; + +use std::time::Duration; + +use base64::{engine::general_purpose::STANDARD, Engine as _}; +use buzz_relay_gateway::Limits; +use common::{body_of, limits, setup, sha256_hex, Gateway, Mode}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; + +/// A POST /query frame, for requests whose answer the test reads later. +fn query(id: &str) -> Value { + json!({"type":"http","id":id,"method":"POST","path":"/query","headers":{"authorization":"x"},"body_b64":""}) +} + +/// Deterministic bytes that are not valid UTF-8. +fn noise(len: usize) -> Vec { + let mut x: u32 = 0x9e37_79b9; + (0..len) + .map(|_| { + x ^= x << 13; + x ^= x >> 17; + x ^= x << 5; + (x >> 24) as u8 | 0x80 + }) + .collect() +} + +#[tokio::test] +async fn bodies_and_headers_arrive_untouched() { + let (relay, _gw, mut w) = setup(limits()).await; + let odd_json = b" [ {\"kinds\" :[0] ,\n\t\"limit\":1 } ] "; + for (i, body) in [noise(64 << 10), odd_json.to_vec()].iter().enumerate() { + let res = w.post(&format!("b{i}"), "/events", body).await; + assert_eq!( + (res["type"].as_str(), res["status"].as_u64()), + (Some("http_res"), Some(200)) + ); + let echo: Value = serde_json::from_slice(&body_of(&res)).unwrap(); + assert_eq!(echo["sha256"], sha256_hex(body)); + assert_eq!(echo["len"], body.len()); + let seen = echo["headers"].as_object().unwrap(); + assert_eq!(seen["host"], relay.addr.to_string()); + assert_eq!( + seen["user-agent"], + concat!("buzz-relay-gateway/", env!("CARGO_PKG_VERSION")) + ); + assert_eq!(seen["authorization"], "Nostr test"); + for name in seen.keys() { + let banned = name.starts_with("x-forwarded") + || [ + "forwarded", + "x-real-ip", + "x-pubkey", + "accept-encoding", + "cookie", + "origin", + ] + .contains(&name.as_str()); + assert!(!banned, "{name} reached the relay"); + } + assert_eq!(res["headers"], json!({"content-type": "application/json"})); + } + assert_eq!(relay.hits("/events"), 2); +} + +#[tokio::test] +async fn response_bytes_come_back_identical() { + let (relay, _gw, mut w) = setup(limits()).await; + let bytes = noise(100_000); + relay.set(Mode::Reply(200, bytes.clone())); + let res = w.post("r", "/query", b"[]").await; + assert_eq!(body_of(&res), bytes); + // Only content-type and retry-after come back; set-cookie stays at the gateway. + assert_eq!( + res["headers"], + json!({"content-type": "application/octet-stream"}) + ); +} + +#[tokio::test] +async fn a_signed_nip98_request_still_verifies_at_the_relay() { + let (relay, _gw, mut w) = setup(limits()).await; + let keys = Keys::generate(); + let body = br#"[{"kinds":[9],"limit":5}]"#; + let url = format!("http://{}/query", relay.addr); + let tags = [ + ["u", url.as_str()], + ["method", "POST"], + ["payload", &sha256_hex(body)], + ] + .map(|t| Tag::parse(t).unwrap()); + let event = EventBuilder::new(Kind::HttpAuth, "") + .tags(tags) + .sign_with_keys(&keys) + .unwrap(); + let auth = format!( + "Nostr {}", + STANDARD.encode(serde_json::to_string(&event).unwrap()) + ); + let headers = json!({"authorization": auth, "content-type": "application/json", "accept": "application/json"}); + let res = w.http("n", "POST", "/query", headers, Some(body)).await; + let echo: Value = serde_json::from_slice(&body_of(&res)).unwrap(); + assert_eq!(echo["nip98"], true); +} + +#[tokio::test] +async fn relay_answers_pass_through_once() { + let (relay, _gw, mut w) = setup(limits()).await; + let limited = br#"{"error":"rate-limited: quota exceeded; retry in 12s"}"#.to_vec(); + relay.set(Mode::Reply(429, limited.clone())); + let res = w.post("a", "/query", b"[]").await; + assert_eq!(res["status"], 429); + assert_eq!(body_of(&res), limited); + relay.set(Mode::Reply(500, b"boom".to_vec())); + assert_eq!(w.post("b", "/count", b"[]").await["status"], 500); + relay.set(Mode::CloseMid); + let res = w.post("c", "/events", b"{}").await; + assert_eq!(res["code"], "upstream_connect"); + relay.set(Mode::Redirect); + let res = w.post("d", "/events", b"{}").await; + assert_eq!( + (res["status"].as_u64(), &res["headers"]), + (Some(302), &json!({})) + ); + // Every request reached the relay exactly once, and the redirect was not followed. + assert_eq!(relay.hits("/query"), 1); + assert_eq!(relay.hits("/count"), 1); + assert_eq!(relay.hits("/events"), 2); + assert_eq!(relay.hits("/elsewhere"), 0); +} + +#[tokio::test] +async fn size_limits_hold_both_ways() { + let (relay, _gw, mut w) = setup(limits()).await; + let res = w.post("big", "/events", &vec![b'a'; (1 << 20) + 1]).await; + assert_eq!(res["code"], "body_too_large"); + assert_eq!(relay.hits("/events"), 0); + relay.set(Mode::Reply(200, vec![b'a'; (4 << 20) + 1])); + assert_eq!( + w.post("huge", "/query", b"[]").await["code"], + "response_too_large" + ); +} + +#[tokio::test] +async fn frames_outside_the_table_are_refused() { + let (relay, _gw, mut w) = setup(limits()).await; + let auth = json!({"authorization": "Nostr test"}); + let cases = [ + ( + json!({"type":"http","id":"1","method":"POST","path":"/query?x=1","headers":auth,"body_b64":""}), + "not_allowed", + ), + ( + json!({"type":"http","id":"2","method":"POST","path":"/query","headers":{"authorization":"x","x-pubkey":"00"},"body_b64":""}), + "not_allowed", + ), + ( + json!({"type":"http","id":"3","method":"POST","path":"/query","headers":auth}), + "not_allowed", + ), + ( + json!({"type":"http","id":"4","method":"POST","path":"/query","headers":auth,"body_b64":"!!"}), + "bad_base64", + ), + ( + json!({"type":"http","id":"5","method":"GET","path":"/","headers":{"accept":"text/html"}}), + "not_allowed", + ), + ]; + for (frame, code) in cases { + let id = frame["id"].clone(); + w.send(frame).await; + let res = w.frame().await; + assert_eq!((&res["id"], res["code"].as_str()), (&id, Some(code))); + } + assert_eq!(relay.hits("/query") + relay.hits("/"), 0); + let nip11 = json!({"accept": "application/nostr+json"}); + let info = w.http("i", "GET", "/", nip11, None).await; + let doc: Value = serde_json::from_slice(&body_of(&info)).unwrap(); + assert_eq!(doc["accept"], "application/nostr+json"); +} + +#[tokio::test] +async fn in_flight_cap_duplicates_and_timeouts() { + let quick = Limits { + request_timeout: Duration::from_millis(800), + ..limits() + }; + let (relay, _gw, mut w) = setup(quick).await; + relay.set(Mode::Stall); + for i in 0..16 { + w.send(query(&format!("s{i}"))).await; + } + w.send(query("s0")).await; + assert_eq!(w.frame().await["code"], "duplicate_id"); + assert_eq!(w.post("s16", "/query", b"").await["code"], "busy"); + for _ in 0..16 { + assert_eq!(w.frame().await["code"], "upstream_timeout"); + } + relay.set(Mode::Echo); + assert_eq!(w.post("after", "/query", b"[]").await["status"], 200); +} + +#[tokio::test] +async fn a_handshake_that_never_ends_is_upstream_connect() { + // Takes the TCP connection and never answers, so the TLS handshake stalls. + let silent = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = silent.local_addr().unwrap(); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((tcp, _)) = silent.accept().await { + held.push(tcp); + } + }); + let quick = Limits { + connect_timeout: Duration::from_millis(300), + ..limits() + }; + let gw = Gateway::start(&format!("wss://{addr}"), quick).await; + let mut w = gw.worker().await; + let res = w.post("c", "/query", b"[]").await; + assert_eq!(res["code"], "upstream_connect", "{res}"); +} + +#[tokio::test] +async fn shutdown_drains_in_flight_requests() { + let (relay, gw, mut w) = setup(limits()).await; + relay.set(Mode::Stall); + w.send(query("slow")).await; + tokio::time::sleep(Duration::from_millis(100)).await; + gw.shutdown.cancel(); + let health = reqwest::get(gw.url("/healthz")).await.unwrap(); + assert_eq!(health.status(), 503); + assert_eq!( + w.post("late", "/query", b"[]").await["code"], + "shutting_down" + ); + assert_eq!(w.closed().await, Some(1012)); + tokio::time::timeout(Duration::from_secs(5), gw.task) + .await + .unwrap() + .unwrap() + .unwrap(); +} diff --git a/crates/buzz-relay-gateway/tests/link.rs b/crates/buzz-relay-gateway/tests/link.rs new file mode 100644 index 00000000000..4b5a699bc12 --- /dev/null +++ b/crates/buzz-relay-gateway/tests/link.rs @@ -0,0 +1,185 @@ +mod common; + +use std::time::Duration; + +use buzz_relay_gateway::Limits; +use common::{bearer, limits, Gateway, Worker}; +use futures_util::SinkExt; +use nostr::Keys; +use serde_json::json; +use tokio_tungstenite::tungstenite::{Error, Message}; + +// Nothing listens on port 9: link control never reaches the relay. +const NO_RELAY: &str = "ws://127.0.0.1:9"; + +#[tokio::test] +async fn bearer_gate_is_single_use_and_fails_closed() { + let gw = Gateway::start( + NO_RELAY, + Limits { + replay_entries: 1, + ..limits() + }, + ) + .await; + let http = reqwest::Client::new(); + let link = |auth: Option<&str>| { + let req = http.get(gw.url("/v1/link")); + let req = match auth { + Some(auth) => req.header("authorization", auth), + None => req, + }; + async move { req.send().await.unwrap() } + }; + let none = link(None).await; + assert_eq!(none.status(), 401); + assert_eq!(none.text().await.unwrap(), r#"{"error":"unauthorized"}"#); + assert_eq!(link(Some(&bearer(&Keys::generate()))).await.status(), 401); + // Refused bearers used no replay slot, so the one-slot cache still admits this one; + // it then fails only the upgrade, because this is a plain GET. + let good = bearer(&gw.keys); + assert_eq!(link(Some(&good)).await.status(), 400); + assert_eq!(link(Some(&good)).await.status(), 401); + assert_eq!(link(Some(&bearer(&gw.keys))).await.status(), 503); + + let health = http.get(gw.url("/healthz")).send().await.unwrap(); + assert_eq!(health.status(), 200); + assert_eq!(health.text().await.unwrap(), "ok"); + assert_eq!( + http.get(gw.url("/v1/other")).send().await.unwrap().status(), + 404 + ); + + gw.shutdown.cancel(); + gw.task.await.unwrap().unwrap(); +} + +#[tokio::test] +async fn bad_bearers_get_no_upgrade() { + let gw = Gateway::start(NO_RELAY, limits()).await; + for bearer in [bearer(&Keys::generate()), "Nostr x".to_owned()] { + let Err(Error::Http(response)) = Worker::connect(&gw, &bearer).await else { + panic!("upgraded with a bad bearer"); + }; + assert_eq!(response.status(), 401); + } +} + +#[tokio::test] +async fn hello_then_ping_pong() { + let gw = Gateway::start(NO_RELAY, limits()).await; + let mut w = gw.worker().await; + let hello = &w.hello; + assert_eq!(hello["v"], 1); + assert_eq!( + hello["expires_at"].as_u64().unwrap() - hello["rotate_at"].as_u64().unwrap(), + 300 + ); + assert_eq!( + hello["limits"], + json!({"max_request_body":1048576,"max_response_body":4194304,"max_ws_text":524288, + "max_http_inflight":16,"max_sockets":8}) + ); + w.send(json!({"type":"ping","ts":1793149200123u64})).await; + assert_eq!( + w.frame().await, + json!({"type":"pong","ts":1793149200123u64}) + ); + w.send(json!({"type":"later","x":1})).await; + assert_eq!(w.frame().await["code"], "unknown_type"); + w.send(json!({"type":"ping"})).await; + assert_eq!(w.frame().await, json!({"type":"pong"})); +} + +#[tokio::test] +async fn gateway_pings_and_silence_closes_4002() { + let quick = Limits { + ping_every: Duration::from_millis(100), + idle: Duration::from_millis(600), + ..limits() + }; + let gw = Gateway::start(NO_RELAY, quick).await; + let mut w = gw.worker().await; + let ping = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Some(Ok(Message::Text(text))) = futures_util::StreamExt::next(&mut w.ws).await { + if text.contains(r#""type":"ping""#) { + return; + } + } + } + }); + ping.await.unwrap(); + assert_eq!(w.closed().await, Some(4002)); +} + +#[tokio::test] +async fn rotate_then_close_4001() { + let short = Limits { + link_lifetime: Duration::from_secs(2), + rotate_before: Duration::from_secs(1), + ..limits() + }; + let gw = Gateway::start(NO_RELAY, short).await; + let mut w = gw.worker().await; + let rotate = w.frame().await; + assert_eq!(rotate["type"], "rotate"); + assert_eq!(rotate["close_at"], w.hello["expires_at"]); + assert_eq!(w.closed().await, Some(4001)); +} + +#[tokio::test] +async fn fifth_link_gets_503() { + let gw = Gateway::start(NO_RELAY, limits()).await; + let mut links = Vec::new(); + for _ in 0..4 { + links.push(gw.worker().await); + } + let Err(Error::Http(response)) = Worker::connect(&gw, &bearer(&gw.keys)).await else { + panic!("a fifth link opened"); + }; + assert_eq!(response.status(), 503); + links.pop().unwrap().ws.close(None).await.unwrap(); + // The freed slot is usable again once that link has torn down. + tokio::time::sleep(Duration::from_millis(300)).await; + gw.worker().await; +} + +#[tokio::test] +async fn protocol_violations_close_the_link() { + let gw = Gateway::start( + NO_RELAY, + Limits { + max_frame: 1024, + ..limits() + }, + ) + .await; + let mut w = gw.worker().await; + w.ws.send(Message::Binary(vec![1, 2, 3].into())) + .await + .unwrap(); + assert_eq!(w.closed().await, Some(1003)); + + let mut w = gw.worker().await; + w.ws.send(Message::text("not json")).await.unwrap(); + assert_eq!(w.frame().await["code"], "bad_frame"); + assert_eq!(w.closed().await, Some(1008)); + + let mut w = gw.worker().await; + w.ws.send(Message::text("x".repeat(2048))).await.unwrap(); + assert_eq!(w.closed().await, Some(1009)); +} + +#[tokio::test] +async fn shutdown_closes_links_1012_and_serve_returns() { + let gw = Gateway::start(NO_RELAY, limits()).await; + let mut w = gw.worker().await; + gw.shutdown.cancel(); + assert_eq!(w.closed().await, Some(1012)); + tokio::time::timeout(Duration::from_secs(5), gw.task) + .await + .unwrap() + .unwrap() + .unwrap(); +} diff --git a/crates/buzz-relay-gateway/tests/logs.rs b/crates/buzz-relay-gateway/tests/logs.rs new file mode 100644 index 00000000000..f3558172d63 --- /dev/null +++ b/crates/buzz-relay-gateway/tests/logs.rs @@ -0,0 +1,72 @@ +//! Its own test binary, because it installs the global subscriber the way +//! main.rs does: fmt().init() also bridges `log` records, which is how +//! tungstenite and reqwest log. +mod common; + +use std::{ + io::Write, + sync::{Arc, Mutex}, +}; + +use common::{bearer_with, limits, Gateway, Relay, Worker}; +use nostr::{EventBuilder, Keys, RelayUrl}; +use serde_json::json; + +#[derive(Clone, Default)] +struct Capture(Arc>>); + +impl Write for Capture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[tokio::test] +async fn logs_never_carry_bearers_bodies_or_frames() { + const MARK: &str = "MARK7f3a"; + let capture = Capture::default(); + let writer = capture.clone(); + // RUST_LOG=trace: the most anyone can turn on. + tracing_subscriber::fmt() + .json() + .with_env_filter(buzz_relay_gateway::log_filter("trace")) + .with_writer(move || writer.clone()) + .init(); + + let relay = Relay::start().await; + let gw = Gateway::start(&relay.ws_url(), limits()).await; + let mut w = Worker::connect(&gw, &bearer_with(&gw.keys, MARK)) + .await + .unwrap(); + let res = w + .http( + "h", + "POST", + "/query", + json!({"authorization": format!("Nostr {MARK}")}), + Some(MARK.as_bytes()), + ) + .await; + assert_eq!(res["status"], 200); + let challenge = w.open("s").await; + let url = RelayUrl::parse(&relay.ws_url()).unwrap(); + let event = EventBuilder::auth(challenge, url) + .sign_with_keys(&Keys::generate()) + .unwrap(); + w.say("s", json!(["AUTH", event, MARK])).await; + w.say("s", json!(["REQ", MARK])).await; + w.send(json!({"type":"ws_close","sid":"s","reason":MARK})) + .await; + w.frame().await; + + let logs = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap(); + assert!(logs.contains("auth_reply_ms"), "the AUTH timing was logged"); + // reqwest logs through `log`, so this line shows the bridge is live. + assert!(logs.contains("starting new connection"), "no `log` records"); + assert!(!logs.contains(MARK), "marker leaked into logs"); + assert!(!logs.contains("Nostr "), "a bearer leaked into logs"); +} diff --git a/crates/buzz-relay-gateway/tests/serve.rs b/crates/buzz-relay-gateway/tests/serve.rs new file mode 100644 index 00000000000..f682f365bc0 --- /dev/null +++ b/crates/buzz-relay-gateway/tests/serve.rs @@ -0,0 +1,21 @@ +mod common; + +use std::time::Duration; + +use buzz_relay_gateway::Limits; +use common::Gateway; +use tokio::{io::AsyncWriteExt, net::TcpStream, time::timeout}; + +#[tokio::test] +async fn a_half_sent_request_does_not_hold_up_shutdown() { + let gw = Gateway::start("ws://127.0.0.1:9", Limits::default()).await; + let mut tcp = TcpStream::connect(gw.addr).await.unwrap(); + // No blank line after the headers: hyper keeps waiting for the rest. + tcp.write_all(b"GET /healthz HTTP/1.1\r\nhost: x\r\n") + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + gw.shutdown.cancel(); + let served = timeout(Duration::from_secs(5), gw.task).await; + assert!(matches!(served, Ok(Ok(Ok(())))), "serve did not return"); +} diff --git a/crates/buzz-relay-gateway/tests/ws.rs b/crates/buzz-relay-gateway/tests/ws.rs new file mode 100644 index 00000000000..71f246e1c5a --- /dev/null +++ b/crates/buzz-relay-gateway/tests/ws.rs @@ -0,0 +1,194 @@ +mod common; + +use std::time::Duration; + +use buzz_relay_gateway::Limits; +use common::{limits, setup, Relay, Worker}; +use nostr::{EventBuilder, Keys, RelayUrl}; +use serde_json::json; + +const AUDIO: &str = "/huddle/6f1c2c1e-4b9a-4c71-9a55-0f1d8c3a2b10/audio"; + +async fn authenticate(w: &mut Worker, relay: &Relay, sid: &str) { + let challenge = w.open(sid).await; + let url = RelayUrl::parse(&relay.ws_url()).unwrap(); + let event = EventBuilder::auth(challenge, url) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let ok = w.say(sid, json!(["AUTH", event])).await; + assert_eq!(ok, json!(["OK", event.id, true, ""])); +} + +async fn wait_for(what: &str, check: impl Fn() -> bool) { + for _ in 0..100 { + if check() { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("timed out waiting for {what}"); +} + +#[tokio::test] +async fn nip42_auth_and_subscriptions_pass_through() { + let (relay, _gw, mut w) = setup(limits()).await; + authenticate(&mut w, &relay, "s1").await; + let eose = w.say("s1", json!(["REQ", "sub-1", {"kinds": [9]}])).await; + assert_eq!(eose, json!(["EOSE", "sub-1"])); +} + +#[tokio::test] +async fn closes_reach_the_other_side() { + let (relay, _gw, mut w) = setup(limits()).await; + w.open("r").await; + w.ws_send("r", r#"["CLOSE_WITH",1012]"#).await; + assert_eq!( + w.frame().await, + json!({"type":"ws_closed","sid":"r","by":"relay","code":1012,"reason":"relay restarting"}) + ); + // RFC 6455 5.5.1: the relay gets its close echoed back. + wait_for("the close echo", || { + relay.socket_closes().contains(&Some(1012)) + }) + .await; + w.open("w").await; + w.send(json!({"type":"ws_close","sid":"w","code":4000,"reason":"done"})) + .await; + assert_eq!( + w.frame().await, + json!({"type":"ws_closed","sid":"w","by":"worker","code":4000,"reason":"done"}) + ); + wait_for("close 4000", || relay.socket_closes().contains(&Some(4000))).await; + // A closed socket is gone, and its sid can be opened again. + w.ws_send("w", "[]").await; + assert_eq!(w.frame().await["code"], "unknown_socket"); + tokio::time::sleep(Duration::from_millis(100)).await; + w.open("w").await; +} + +#[tokio::test] +async fn closing_the_link_closes_every_relay_socket() { + let (relay, _gw, mut w) = setup(limits()).await; + for sid in ["a", "b", "c"] { + w.open(sid).await; + } + assert_eq!(relay.open_sockets(), 3); + w.ws.close(None).await.unwrap(); + wait_for("every relay socket to close", || relay.open_sockets() == 0).await; +} + +#[tokio::test] +async fn socket_limits() { + let (relay, _gw, mut w) = setup(Limits { + max_ws_text: 64, + ..limits() + }) + .await; + for i in 0..8 { + w.open(&format!("s{i}")).await; + } + w.send(json!({"type":"ws_open","sid":"s8","path":"/"})) + .await; + assert_eq!( + w.frame().await, + json!({"type":"error","sid":"s8","code":"busy","message":"too many requests or sockets in flight"}) + ); + w.send(json!({"type":"ws_open","sid":"s0","path":"/"})) + .await; + assert_eq!(w.frame().await["code"], "duplicate_id"); + w.ws_send("s0", &"x".repeat(65)).await; + assert_eq!(w.frame().await["code"], "frame_too_large"); + // The socket stays usable after a refused frame. + assert_eq!(w.say("s0", json!(["REQ", "q"])).await, json!(["EOSE", "q"])); + for (path, code) in [(AUDIO, "not_supported"), ("/x", "not_allowed")] { + w.send(json!({"type":"ws_close","sid":"s7"})).await; + assert_eq!(w.frame().await["by"], "worker"); + tokio::time::sleep(Duration::from_millis(100)).await; + w.send(json!({"type":"ws_open","sid":"s7","path":path})) + .await; + assert_eq!(w.frame().await["code"], code); + w.open("s7").await; + } + assert_eq!(relay.hits("/x"), 0); +} + +#[tokio::test] +async fn refused_and_failed_opens_report_ws_closed() { + let (relay, _gw, mut w) = setup(limits()).await; + relay.set(common::Mode::Reply(503, b"busy".to_vec())); + w.send(json!({"type":"ws_open","sid":"x","path":"/"})).await; + assert_eq!( + w.frame().await, + json!({"type":"ws_closed","sid":"x","by":"gateway","code":null,"reason":"upgrade_refused","http_status":503}) + ); +} + +#[tokio::test] +async fn relay_side_limits_close_one_socket() { + let quick = Limits { + max_relay_frame: 1024, + relay_silence: Duration::from_millis(500), + ..limits() + }; + let (_relay, _gw, mut w) = setup(quick).await; + w.open("big").await; + w.ws_send("big", r#"["SEND",1,2048]"#).await; + let closed = w.frame().await; + assert_eq!(closed["by"], "gateway"); + assert_eq!(closed["reason"], "frame_too_large"); + w.open("quiet").await; + w.ws_send("quiet", r#"["SILENT"]"#).await; + assert_eq!(w.frame().await["reason"], "relay_silent"); +} + +#[tokio::test] +async fn a_relay_that_stops_reading_closes_that_socket() { + // A queue that never fills, so only the silence timer can end the socket. + let stalled = Limits { + relay_queue: 1024, + relay_silence: Duration::from_secs(5), + ..limits() + }; + let (_relay, _gw, mut w) = setup(stalled).await; + w.open("stuck").await; + // The mock relay stops reading and sending; the write stalls once the TCP buffers fill. + w.ws_send("stuck", r#"["SILENT"]"#).await; + let big = "x".repeat(256 << 10); + for _ in 0..100 { + w.ws_send("stuck", &big).await; + } + let closed = w.frame().await; + assert_eq!( + (closed["type"].as_str(), closed["reason"].as_str()), + (Some("ws_closed"), Some("relay_silent")), + "{closed}" + ); +} + +#[tokio::test] +async fn a_worker_that_stops_reading_loses_the_link() { + let slow = Limits { + out_frames: 16, + out_bytes: 64 << 10, + ..limits() + }; + let (relay, _gw, mut w) = setup(slow).await; + w.open("f").await; + w.ws_send("f", r#"["SEND",4000,4096]"#).await; + // Stop reading while the relay floods; the queue fills behind the full TCP buffers. + tokio::time::sleep(Duration::from_millis(500)).await; + assert_eq!(w.closed().await, Some(4003)); + wait_for("the relay socket to close", || relay.open_sockets() == 0).await; +} + +#[tokio::test] +async fn shutdown_closes_sockets_then_the_link() { + let (relay, gw, mut w) = setup(limits()).await; + w.open("s").await; + gw.shutdown.cancel(); + let closed = w.frame().await; + assert_eq!(closed["by"], "gateway"); + assert_eq!(closed["reason"], "shutdown"); + assert_eq!(w.closed().await, Some(1012)); + wait_for("close 1001", || relay.socket_closes().contains(&Some(1001))).await; +}