diff --git a/Cargo.lock b/Cargo.lock index 870a7493..7590cfee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4721,6 +4721,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tokio-stream", + "tokio-tungstenite", "tokio-util", "toml 1.1.2+spec-1.1.0", "tower", diff --git a/Cargo.toml b/Cargo.toml index 250ec8a5..049f814e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,10 @@ axum = { version = "0.8.8", features = ["multipart", "http1", "http2", "tokio", tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs", "process", "signal"] } tokio-util = { version = "0.7.18", features = ["io", "codec", "compat"] } tokio-stream = { version = "0.1.18", features = ["fs", "sync"] } +# Only pulled in by the `test_utils` feature (rt-hurl-helper bin). Not +# shipped in release; pinned to the same 0.28 axum 0.8.8 already brings +# transitively so we don't duplicate the crate graph. +tokio-tungstenite = { version = "0.28", default-features = false, features = ["connect", "handshake"], optional = true } bytes = "1.11.1" tempfile = "3.27.0" tower = "0.5.3" @@ -190,7 +194,7 @@ metrics-exporter-prometheus = { version = "0.18", default-features = false } [features] default = [] -test_utils = ["mockall"] +test_utils = ["mockall", "dep:tokio-tungstenite"] integration_tests = [] # WASM plugin runtime (Extism). Opt-in: bundles wasmtime, a large engine most # deployments won't use. Activation also requires OXICLOUD_ENABLE_PLUGINS=true. @@ -262,6 +266,16 @@ path = "src/bin/generate-openapi.rs" # Invoked by `just openapi`, which passes `--features dev_tools`. required-features = ["dev_tools"] +[[bin]] +name = "generate-asyncapi" +path = "src/bin/generate-asyncapi.rs" +# Dev-only: regenerates `resources/gen/asyncapi.json` — the WS surface's +# analogue of openapi.json. Constructed from the same `error_code` +# constants + `Topic`/`RealtimeEvent` shapes the server uses, so the +# spec stays in sync with the implementation by construction. Same +# gating as `generate-openapi`. Invoked by `just asyncapi`. +required-features = ["dev_tools"] + [[bin]] name = "opaque-hurl-helper" path = "src/bin/opaque-hurl-helper.rs" @@ -292,6 +306,21 @@ path = "src/bin/load-seed.rs" # and load-nightly.yml build it explicitly with --features load_seed_bin. required-features = ["load_seed_bin"] +[[bin]] +name = "rt-hurl-helper" +path = "src/bin/rt-hurl-helper.rs" +# Test-suite WebSocket client for the realtime message bus. Hurl is +# HTTP-only and cannot drive a WS handshake or read frames; this bin +# supplies the two modes the smoke test needs — `subscribe-and-collect` +# (background subscriber that captures events to JSON) and +# `expect-denied` (synchronous check that a subscribe attempt is +# rejected with a specific JSON-RPC error code). Invoked from +# tests/api/rt_bus_check.sh after the main hurl block. +# +# Not shipped in release: gated behind `test_utils` alongside the +# opaque/dpop helpers. +required-features = ["test_utils"] + # Phase 0 perf harness — Task 0.2 (criterion latency + output-size bench). [[bench]] name = "thumbnails" diff --git a/justfile b/justfile index fe306d35..39157f38 100644 --- a/justfile +++ b/justfile @@ -194,6 +194,13 @@ audit: openapi: cargo run --features dev_tools --bin generate-openapi +# Regenerate `resources/gen/asyncapi.json` — the WS surface's spec, +# analogue of openapi.json. Built from the `Topic`, `RealtimeEvent`, +# and `error_code` constants in `application/ports/realtime_ports.rs` +# so the spec stays in sync with the wire by construction. +asyncapi: + cargo run --features dev_tools --bin generate-asyncapi + db: docker compose up -d postgres diff --git a/src/bin/opaque-hurl-helper.rs b/src/bin/opaque-hurl-helper.rs index b7f2d974..ed8325f4 100644 --- a/src/bin/opaque-hurl-helper.rs +++ b/src/bin/opaque-hurl-helper.rs @@ -402,8 +402,79 @@ async fn main() -> ExitCode { Err(e) => return fail(format!("/api/admin/sessions network: {e}")), } + // ── OPAQUE-minted JWT works against the realtime WS ───────────── + // + // Regression guard: `auth_middleware` doesn't inspect how a JWT + // was minted, so an OPAQUE-issued access_token must Just Work on + // `/api/rt/ws` the same way a legacy-password one does. If a + // future refactor makes WS auth diverge from the general + // request-auth path, this smoke fails and the divergence gets + // caught here rather than only surfacing in the collab editor. + // + // The check itself is trivial: connect with the OPAQUE JWT, send + // one `rt.ping`, expect `result.pong == true`. + if let Err(msg) = opaque_jwt_ws_smoke(base, &auth.access_token).await { + return fail(format!("OPAQUE JWT + WS: {msg}")); + } + eprintln!( - "opaque-hurl-helper: OK — register + login + /me + admin sessions origin=opaque for '{username}'" + "opaque-hurl-helper: OK — register + login + /me + admin sessions origin=opaque + rt.ping over WS for '{username}'" ); ExitCode::from(EXIT_OK) } + +async fn opaque_jwt_ws_smoke(base: &str, access_token: &str) -> Result<(), String> { + use futures::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message; + use tokio_tungstenite::tungstenite::client::IntoClientRequest; + use tokio_tungstenite::tungstenite::http::HeaderValue; + + let ws_url = match base.strip_prefix("http://") { + Some(rest) => format!("ws://{rest}/api/rt/ws"), + None => match base.strip_prefix("https://") { + Some(rest) => format!("wss://{rest}/api/rt/ws"), + None => return Err(format!("unexpected base scheme: {base}")), + }, + }; + + let mut req = ws_url + .into_client_request() + .map_err(|e| format!("bad url: {e}"))?; + req.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&format!("Bearer {access_token}")) + .map_err(|e| format!("bad bearer header: {e}"))?, + ); + let (mut ws, _resp) = tokio_tungstenite::connect_async(req) + .await + .map_err(|e| format!("connect failed: {e}"))?; + + let ping = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "rt.ping", + }) + .to_string(); + ws.send(Message::Text(ping.into())) + .await + .map_err(|e| format!("send: {e}"))?; + + // Bounded wait — the server should reply immediately. A hung reply + // means the WS handler didn't recognise the JWT (misgated + // middleware) or panicked; we treat either as a hard failure. + let msg = tokio::time::timeout(std::time::Duration::from_secs(2), ws.next()) + .await + .map_err(|_| "rt.ping response timed out".to_string())? + .ok_or_else(|| "socket closed before response".to_string())? + .map_err(|e| format!("recv: {e}"))?; + + let Message::Text(text) = msg else { + return Err(format!("expected text frame, got {msg:?}")); + }; + let v: serde_json::Value = + serde_json::from_str(&text).map_err(|e| format!("bad json: {e}: {text}"))?; + if v["result"]["pong"] != true { + return Err(format!("expected pong=true, got: {v}")); + } + Ok(()) +} diff --git a/src/bin/rt-hurl-helper.rs b/src/bin/rt-hurl-helper.rs new file mode 100644 index 00000000..62cf43d4 --- /dev/null +++ b/src/bin/rt-hurl-helper.rs @@ -0,0 +1,426 @@ +//! WebSocket-side smoke-test helper for the realtime message bus. +//! +//! Hurl is HTTP-only — it can't do a WS upgrade, let alone read frames +//! for later assertion. This binary is the WS half of the smoke test: +//! opens `/api/rt/ws`, speaks JSON-RPC 2.0, and either collects events +//! into a JSON file for shell assertions (`subscribe-and-collect`) or +//! validates that an authz-denied subscribe returns the expected wire +//! error code (`expect-denied`). +//! +//! Invocation (from `tests/api/rt_bus_check.sh`): +//! +//! ```bash +//! rt-hurl-helper subscribe-and-collect \ +//! --url ws://127.0.0.1:$PORT/api/rt/ws \ +//! --token $USER_JWT \ +//! --subscribe folder:$FOLDER_A \ +//! --expect-events 1 \ +//! --timeout 3s \ +//! --output /tmp/rt_s1.json & +//! +//! rt-hurl-helper expect-denied \ +//! --url ws://127.0.0.1:$PORT/api/rt/ws \ +//! --token $USER2_JWT \ +//! --subscribe folder:$FOLDER_A \ +//! --reason no_read \ +//! --timeout 2s +//! ``` +//! +//! Exit codes: +//! * 0 — expectation met. +//! * 1 — expectation failed (wrong event, unexpected event, timeout +//! without hitting the target, denied when expecting event, +//! event when expecting denied). +//! * 2 — protocol / connect error the shell can distinguish from a +//! real assertion failure. +//! +//! JSON output shape for `subscribe-and-collect` (written to `--output`): +//! +//! ```jsonc +//! { +//! "subscribed": ["folder:..."], +//! "events": [ { "topic": "folder:...", "event": "file_created", +//! "data": { ... } } ], +//! "timed_out": false +//! } +//! ``` + +use std::process::ExitCode; +use std::time::Duration; + +use futures::{SinkExt, StreamExt}; +use serde_json::{Value, json}; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::HeaderValue; + +// ════════════════════════════════════════════════════════════════════════════ +// CLI parsing (minimal, dependency-free) +// ════════════════════════════════════════════════════════════════════════════ + +struct Args { + mode: Mode, + url: String, + token: String, + subscribe: Vec, + expect_events: Option, + reason: Option, + timeout: Duration, + output: Option, +} + +enum Mode { + SubscribeAndCollect, + ExpectDenied, +} + +fn parse_duration(s: &str) -> Result { + // Accept `s`, `ms`, or a bare integer (interpreted as + // seconds). Kept small — hurl and shell are the only callers. + let s = s.trim(); + if let Some(num) = s.strip_suffix("ms") { + num.parse::() + .map(Duration::from_millis) + .map_err(|_| format!("bad duration: {s}")) + } else if let Some(num) = s.strip_suffix('s') { + num.parse::() + .map(Duration::from_secs) + .map_err(|_| format!("bad duration: {s}")) + } else { + s.parse::() + .map(Duration::from_secs) + .map_err(|_| format!("bad duration: {s}")) + } +} + +fn parse_args() -> Result { + let mut it = std::env::args().skip(1); + let mode = match it.next().as_deref() { + Some("subscribe-and-collect") => Mode::SubscribeAndCollect, + Some("expect-denied") => Mode::ExpectDenied, + Some(other) => return Err(format!("unknown mode: {other}")), + None => return Err("mode is required".into()), + }; + + let mut url = None; + let mut token = None; + let mut subscribe = Vec::new(); + let mut expect_events = None; + let mut reason = None; + let mut timeout = Duration::from_secs(3); + let mut output = None; + + while let Some(flag) = it.next() { + let value = it + .next() + .ok_or_else(|| format!("flag {flag} requires a value"))?; + match flag.as_str() { + "--url" => url = Some(value), + "--token" => token = Some(value), + "--subscribe" => subscribe.push(value), + "--expect-events" => { + expect_events = Some( + value + .parse::() + .map_err(|_| format!("--expect-events not a number: {value}"))?, + ); + } + "--reason" => reason = Some(value), + "--timeout" => timeout = parse_duration(&value)?, + "--output" => output = Some(value), + other => return Err(format!("unknown flag: {other}")), + } + } + + Ok(Args { + mode, + url: url.ok_or("--url required")?, + token: token.ok_or("--token required")?, + subscribe, + expect_events, + reason, + timeout, + output, + }) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Main +// ════════════════════════════════════════════════════════════════════════════ + +#[tokio::main(flavor = "current_thread")] +async fn main() -> ExitCode { + let args = match parse_args() { + Ok(a) => a, + Err(e) => { + eprintln!("rt-hurl-helper: {e}"); + return ExitCode::from(2); + } + }; + + let result = match args.mode { + Mode::SubscribeAndCollect => subscribe_and_collect(args).await, + Mode::ExpectDenied => expect_denied(args).await, + }; + + match result { + Ok(()) => ExitCode::SUCCESS, + Err(HelperError::Expectation(msg)) => { + eprintln!("rt-hurl-helper: expectation failed: {msg}"); + ExitCode::from(1) + } + Err(HelperError::Protocol(msg)) => { + eprintln!("rt-hurl-helper: protocol error: {msg}"); + ExitCode::from(2) + } + } +} + +// ════════════════════════════════════════════════════════════════════════════ +// Errors +// ════════════════════════════════════════════════════════════════════════════ + +enum HelperError { + /// The wire behaved OK but didn't match what the test expected — + /// e.g. a `subscribed` ack when we expected `denied`, or fewer + /// events than requested before timeout. Exit 1: test failure. + Expectation(String), + /// Something is broken at the transport/JSON layer — connect + /// refused, malformed frame, TLS handshake failed. Exit 2: + /// infrastructure problem, not a test result. + Protocol(String), +} + +impl From for HelperError { + fn from(e: E) -> Self { + HelperError::Protocol(e.to_string()) + } +} + +// ════════════════════════════════════════════════════════════════════════════ +// WS connection +// ════════════════════════════════════════════════════════════════════════════ + +/// Open a WS connection to `url` with the given bearer token attached +/// via `Authorization: Bearer `. Programmatic client — this is the +/// path native clients (this helper, future sync-client integrations) +/// take. Browser clients that can't set the header will use the +/// `Sec-WebSocket-Protocol` subprotocol fallback (Phase A follow-up). +async fn connect_ws( + url: &str, + token: &str, +) -> Result< + tokio_tungstenite::WebSocketStream>, + HelperError, +> { + let mut req = url + .into_client_request() + .map_err(|e| HelperError::Protocol(format!("bad url: {e}")))?; + let bearer = format!("Bearer {token}"); + req.headers_mut().insert( + "Authorization", + HeaderValue::from_str(&bearer) + .map_err(|e| HelperError::Protocol(format!("bad token: {e}")))?, + ); + let (ws, _resp) = tokio_tungstenite::connect_async(req) + .await + .map_err(|e| HelperError::Protocol(format!("connect failed: {e}")))?; + Ok(ws) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Mode: subscribe-and-collect +// ════════════════════════════════════════════════════════════════════════════ + +async fn subscribe_and_collect(args: Args) -> Result<(), HelperError> { + if args.subscribe.is_empty() { + return Err(HelperError::Protocol( + "--subscribe required for subscribe-and-collect".into(), + )); + } + let expect_events = args.expect_events.unwrap_or(0); + + let mut ws = connect_ws(&args.url, &args.token).await?; + + // Subscribe to every requested topic; track pending request ids so + // we know when all acks have arrived before we start counting + // events. + let mut subscribed: Vec = Vec::new(); + let mut pending_subs: std::collections::HashMap = std::collections::HashMap::new(); + for (i, topic) in args.subscribe.iter().enumerate() { + let req_id = (i as u64) + 1; + let frame = json!({ + "jsonrpc": "2.0", + "id": req_id, + "method": "rt.subscribe", + "params": { "topic": topic }, + }); + ws.send(Message::Text(frame.to_string().into())).await?; + pending_subs.insert(req_id, topic.clone()); + } + + let mut events: Vec = Vec::new(); + let mut timed_out = false; + + let deadline = tokio::time::Instant::now() + args.timeout; + + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + timed_out = true; + break; + } + // Exit early: all acks received AND enough events collected. + if pending_subs.is_empty() && events.len() >= expect_events { + break; + } + + let msg = match timeout(remaining, ws.next()).await { + Ok(Some(Ok(m))) => m, + Ok(Some(Err(e))) => { + return Err(HelperError::Protocol(format!("ws error: {e}"))); + } + Ok(None) => { + return Err(HelperError::Protocol("connection closed by peer".into())); + } + Err(_) => { + timed_out = true; + break; + } + }; + + let Message::Text(text) = msg else { + // Ignore ping/pong/binary; server may send close later. + continue; + }; + let value: Value = serde_json::from_str(&text) + .map_err(|e| HelperError::Protocol(format!("bad frame: {e}: {text}")))?; + + // Response to a subscribe request? + if let Some(id_num) = value.get("id").and_then(|v| v.as_u64()) { + let topic = pending_subs.remove(&id_num); + if let Some(err) = value.get("error") { + return Err(HelperError::Expectation(format!( + "subscribe to {} denied: {}", + topic.as_deref().unwrap_or(""), + err, + ))); + } + if let Some(topic) = topic { + subscribed.push(topic); + } + continue; + } + + // Notification (id-less)? + let method = value.get("method").and_then(|v| v.as_str()).unwrap_or(""); + if method == "rt.event" + && let Some(params) = value.get("params") + { + events.push(params.clone()); + } + // Other notifications (`rt.revoked`, `rt.pong`) — ignored for + // subscribe-and-collect. They can be added to the output + // schema when scenarios need them. + } + + // Assertion: at least `expect_events` collected before timeout. + let met = events.len() >= expect_events; + + // Always write output (even on failure) so the shell can diff. + if let Some(path) = args.output.as_ref() { + let summary = json!({ + "subscribed": subscribed, + "events": events, + "timed_out": timed_out, + }); + std::fs::write(path, serde_json::to_vec_pretty(&summary).unwrap()) + .map_err(|e| HelperError::Protocol(format!("write output: {e}")))?; + } + + if !met { + return Err(HelperError::Expectation(format!( + "expected {} events, got {} ({}timeout)", + expect_events, + events.len(), + if timed_out { "with " } else { "no " } + ))); + } + Ok(()) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Mode: expect-denied +// ════════════════════════════════════════════════════════════════════════════ + +async fn expect_denied(args: Args) -> Result<(), HelperError> { + let topic = args + .subscribe + .first() + .ok_or_else(|| HelperError::Protocol("--subscribe required for expect-denied".into()))? + .clone(); + + let mut ws = connect_ws(&args.url, &args.token).await?; + + let req_id: u64 = 1; + let frame = json!({ + "jsonrpc": "2.0", + "id": req_id, + "method": "rt.subscribe", + "params": { "topic": topic }, + }); + ws.send(Message::Text(frame.to_string().into())).await?; + + // Wait for the id-matched response with an `error` object. Any + // notification arriving before the response is skipped — the + // server should not fan out to a subscription that hasn't been + // acked yet, but the check is robust to that ordering anyway. + let deadline = tokio::time::Instant::now() + args.timeout; + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + return Err(HelperError::Expectation( + "timeout without a subscribe response".into(), + )); + } + + let msg = match timeout(remaining, ws.next()).await { + Ok(Some(Ok(m))) => m, + Ok(Some(Err(e))) => return Err(HelperError::Protocol(format!("ws error: {e}"))), + Ok(None) => return Err(HelperError::Protocol("connection closed by peer".into())), + Err(_) => { + return Err(HelperError::Expectation( + "timeout without a subscribe response".into(), + )); + } + }; + let Message::Text(text) = msg else { continue }; + let value: Value = serde_json::from_str(&text) + .map_err(|e| HelperError::Protocol(format!("bad frame: {e}: {text}")))?; + + // Match by id. + let Some(id_num) = value.get("id").and_then(|v| v.as_u64()) else { + continue; + }; + if id_num != req_id { + continue; + } + + // Expect: error object present. + let Some(err) = value.get("error") else { + return Err(HelperError::Expectation(format!( + "expected `error` object, got: {value}" + ))); + }; + let message = err.get("message").and_then(|v| v.as_str()).unwrap_or(""); + if let Some(want) = args.reason.as_ref() + && message != want + { + return Err(HelperError::Expectation(format!( + "expected reason `{want}`, got `{message}` (full error: {err})" + ))); + } + return Ok(()); + } +} diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh new file mode 100755 index 00000000..75a660c6 --- /dev/null +++ b/tests/api/rt_bus_check.sh @@ -0,0 +1,219 @@ +#!/usr/bin/env bash +# ───────────────────────────────────────────────────────────────────────────── +# Realtime bus smoke test — the parts Hurl can't drive. +# +# Hurl is HTTP-only and cannot open a WebSocket, so the WS half of the test +# runs through `rt-hurl-helper` (a small Rust bin gated on `test_utils`). +# This script orchestrates it against a live oxicloud server: bootstraps +# state with curl, exercises the bus, asserts on the helper's JSON output. +# +# Four scenarios: +# S1 Positive delivery — subscribe to folder A, upload into A, see event. +# S2 Topic isolation — subscribe to folder A only, upload into B and +# then A; must see A's event only. +# S3 AuthZ denial — user2 subscribes to folder A owned by user1 +# without a grant; expect wire reason `no_read`. +# S4 Anti-enumeration — subscribe to a folder that does not exist; +# must return the SAME wire reason (`no_read`) +# as S3, per the plan's anti-enum invariant. +# +# Exit non-zero on any failure — run.sh treats that as a suite failure. +# ───────────────────────────────────────────────────────────────────────────── + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +BUILD_TARGET="${BUILD_TARGET:-debug}" +HELPER_BIN="$REPO_ROOT/target/$BUILD_TARGET/rt-hurl-helper" + +# ── Env from test.env (base_url, admin username/password) ──────────────────── +# shellcheck disable=SC1091 +source "$SCRIPT_DIR/test.env" +: "${base_url:?}" "${username:?}" "${password:?}" + +# WS URL derived from base_url (test.env uses http://); tolerate https for +# future deployments even though the test suite runs plain HTTP. +case "$base_url" in + http://*) ws_url="ws://${base_url#http://}/api/rt/ws" ;; + https://*) ws_url="wss://${base_url#https://}/api/rt/ws" ;; + *) echo "rt_bus_check: unexpected base_url scheme: $base_url" >&2; exit 2 ;; +esac + +log() { printf '\033[1;36m[rt_bus_check]\033[0m %s\n' "$*"; } +die() { printf '\033[1;31m[rt_bus_check FAIL]\033[0m %s\n' "$*" >&2; exit 1; } + +# ── Build the helper on demand (matches opaque/dpop helper convention) ────── +if [[ ! -x "$HELPER_BIN" ]]; then + log "Building rt-hurl-helper ($BUILD_TARGET)..." + case "$BUILD_TARGET" in + debug) (cd "$REPO_ROOT" && cargo build --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; + release) (cd "$REPO_ROOT" && cargo build --release --features test_utils --bin rt-hurl-helper 2>&1 | tail -n 20) || die "rt-hurl-helper build failed" ;; + esac +fi + +# ── curl wrappers ─────────────────────────────────────────────────────────── +c_post() { + local url="$1" auth="$2" body="$3" + curl -sS -X POST -H "Content-Type: application/json" \ + ${auth:+-H "Authorization: Bearer $auth"} \ + -d "$body" "$url" +} + +c_get() { + local url="$1" auth="$2" + curl -sS -H "Accept: application/json" \ + ${auth:+-H "Authorization: Bearer $auth"} \ + "$url" +} + +# ── Setup: register fresh users; the test.env admin may be OPAQUE- +# migrated and the legacy password-login path refuses those accounts, +# so we don't use it at all — same pattern as `dedup_admin_gate.hurl` +# and the other hurl scenarios that need a self-contained principal. +# The scenarios only need "a user with their own folders", not admin +# rights. +suffix="$(date +%s)_$$" +user1_name="rtbus_u1_$suffix" +user1_pass="RtBusU1Pass1!" +user2_name="rtbus_u2_$suffix" +user2_pass="RtBusU2Pass1!" + +log "Register user1 ($user1_name) and log in..." +c_post "$base_url/api/auth/register" "" \ + "$(printf '{"username":"%s","email":"%s@example.com","password":"%s"}' \ + "$user1_name" "$user1_name" "$user1_pass")" > /dev/null +u1_login=$(c_post "$base_url/api/auth/login" "" \ + "$(printf '{"username":"%s","password":"%s"}' "$user1_name" "$user1_pass")") +user1_token=$(printf '%s' "$u1_login" | jq -r '.access_token') +[[ -n "$user1_token" && "$user1_token" != "null" ]] || die "no user1 token: $u1_login" + +log "Discover user1 root folder..." +folders=$(c_get "$base_url/api/folders" "$user1_token") +root_id=$(printf '%s' "$folders" | jq -r '.[0].id') +[[ -n "$root_id" && "$root_id" != "null" ]] || die "no root folder for user1: $folders" + +log "Create folder A (rt_bus_A_$suffix) and folder B (rt_bus_B_$suffix)..." +folder_a=$(c_post "$base_url/api/folders" "$user1_token" \ + "$(printf '{"name":"rt_bus_A_%s","parent_id":"%s"}' "$suffix" "$root_id")" | jq -r '.id') +folder_b=$(c_post "$base_url/api/folders" "$user1_token" \ + "$(printf '{"name":"rt_bus_B_%s","parent_id":"%s"}' "$suffix" "$root_id")" | jq -r '.id') +[[ -n "$folder_a" && "$folder_a" != "null" ]] || die "folder A creation failed" +[[ -n "$folder_b" && "$folder_b" != "null" ]] || die "folder B creation failed" + +# Register a second user for the AuthZ-denial scenario (S3). No grant on +# folder A → subscribe attempt must be denied. +log "Register user2 ($user2_name) and log in..." +c_post "$base_url/api/auth/register" "" \ + "$(printf '{"username":"%s","email":"%s@example.com","password":"%s"}' \ + "$user2_name" "$user2_name" "$user2_pass")" > /dev/null +user2_login=$(c_post "$base_url/api/auth/login" "" \ + "$(printf '{"username":"%s","password":"%s"}' "$user2_name" "$user2_pass")") +user2_token=$(printf '%s' "$user2_login" | jq -r '.access_token') +[[ -n "$user2_token" && "$user2_token" != "null" ]] || die "no user2 token: $user2_login" + +# ── Helper: create a small file inside a folder via the byte-upload path. +# Not delta / instant-upload; keeps the wire simple and hits the same +# `upload_file_streaming` publish hook. +mkfile_in() { + local folder_id="$1" name="$2" token="$3" + local tmpfile + tmpfile="$(mktemp -t rtbus_body.XXXXXX)" + printf 'rt-bus-test-payload' > "$tmpfile" + # Multipart-upload path used by the frontend for byte uploads. + # NOTE: `folder_id` MUST come BEFORE the `file` part — file_handler.rs + # streams the parts in order and the fail-fast folder-required check + # fires the moment it sees the file bytes; a folder_id sent after the + # file arrives too late (returns 400 "folder_id is required"). + curl -sS -X POST \ + -H "Authorization: Bearer $token" \ + -F "folder_id=$folder_id" \ + -F "file=@$tmpfile;filename=$name" \ + "$base_url/api/files/upload" > /dev/null + rm -f "$tmpfile" +} + +# ── Scenario 1 — Positive delivery ────────────────────────────────────────── +log "S1: subscribe to folder A, upload into A, expect one file_created event." +out_s1="$(mktemp -t rtbus_s1.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 5s \ + --output "$out_s1" & +helper_pid=$! +# Give the ack a moment to install so the upload's post-commit publish +# lands on a live receiver, not an orphaned map entry. +sleep 0.4 +mkfile_in "$folder_a" "s1.txt" "$user1_token" +if ! wait "$helper_pid"; then + cat "$out_s1" >&2 || true + die "S1: helper did not observe the expected event" +fi +# jq assertions — one event, correct parent_id, correct discriminator. +[[ "$(jq -r '.events | length' "$out_s1")" == "1" ]] \ + || { cat "$out_s1"; die "S1: expected 1 event, got $(jq -r '.events | length' "$out_s1")"; } +[[ "$(jq -r '.events[0].event' "$out_s1")" == "file_created" ]] \ + || die "S1: wrong event discriminator: $(jq -r '.events[0].event' "$out_s1")" +[[ "$(jq -r '.events[0].data.parent_id' "$out_s1")" == "$folder_a" ]] \ + || die "S1: parent_id mismatch" +log "S1 OK" + +# ── Scenario 2 — Topic isolation ──────────────────────────────────────────── +log "S2: subscribe to folder A, upload into B (must be silent) and A (triggers exit)." +out_s2="$(mktemp -t rtbus_s2.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 5s \ + --output "$out_s2" & +helper_pid=$! +sleep 0.4 +# B first — should be dropped for the A subscriber. +mkfile_in "$folder_b" "s2_in_B.txt" "$user1_token" +# Small settle so if isolation is BROKEN, the B event has time to arrive +# before A's; the assertion below then catches it as a wrong parent_id. +sleep 0.2 +mkfile_in "$folder_a" "s2_in_A.txt" "$user1_token" +if ! wait "$helper_pid"; then + cat "$out_s2" >&2 || true + die "S2: helper did not observe the expected A event" +fi +# Exactly one event, and it MUST be from folder A. If isolation were +# broken, we'd either see 2 events or a B-parented event first. +[[ "$(jq -r '.events | length' "$out_s2")" == "1" ]] \ + || { cat "$out_s2"; die "S2: expected 1 event, got $(jq -r '.events | length' "$out_s2") (isolation broken?)"; } +[[ "$(jq -r '.events[0].data.parent_id' "$out_s2")" == "$folder_a" ]] \ + || die "S2: parent_id was $(jq -r '.events[0].data.parent_id' "$out_s2"), expected $folder_a" +log "S2 OK" + +# ── Scenario 3 — AuthZ denial ─────────────────────────────────────────────── +log "S3: user2 subscribes to folder A (no grant); expect no_read denial." +if ! "$HELPER_BIN" expect-denied \ + --url "$ws_url" \ + --token "$user2_token" \ + --subscribe "folder:$folder_a" \ + --reason no_read \ + --timeout 3s; then + die "S3: user2 was NOT denied on folder A (AuthZ gate broken?)" +fi +log "S3 OK" + +# ── Scenario 4 — Anti-enumeration parity ──────────────────────────────────── +log "S4: subscribe to a nonexistent folder; wire reason must equal S3." +fake_folder="00000000-0000-0000-0000-000000000000" +if ! "$HELPER_BIN" expect-denied \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$fake_folder" \ + --reason no_read \ + --timeout 3s; then + die "S4: nonexistent folder did not collapse to no_read (anti-enum invariant broken)" +fi +log "S4 OK" + +log "All four realtime-bus scenarios passed." diff --git a/tests/api/run.sh b/tests/api/run.sh index 7aa66375..5c4e073d 100755 --- a/tests/api/run.sh +++ b/tests/api/run.sh @@ -267,7 +267,19 @@ bash "$API_DIR/thumb_import_check.sh" bash "$API_DIR/storage_cleanup_check.sh" -# ── 5. OPAQUE crypto handshake — the parts Hurl can't drive ───────────── +# ── 5. Realtime message bus — WebSocket smoke test ────────────────────── +# Runs BEFORE the OPAQUE helper so its user registration + login uses +# the legacy password path (opaque_substrate.hurl migrates the admin +# account, but by running first this check is unaffected by whatever +# order later scenarios touch the auth substrate). Four scenarios: +# positive delivery, topic isolation, AuthZ denial on subscribe, +# anti-enumeration parity. See `tests/api/rt_bus_check.sh` and +# `docs/plan/message-bus.md`. +log "Running realtime-bus smoke test..." +BUILD_TARGET="$BUILD_TARGET" bash "$REPO_ROOT/tests/api/rt_bus_check.sh" \ + || die "realtime-bus smoke test failed" + +# ── 6. OPAQUE crypto handshake — the parts Hurl can't drive ───────────── # Full OPAQUE register + login handshake against the running server, # using the real ciphersuite client-side. Closes the gap left by # `opaque_substrate.hurl` (which covers only wire shape, not OPRF- @@ -289,7 +301,7 @@ OPAQUE_HELPER_USERNAME="$username" \ OPAQUE_HELPER_PASSWORD="$password" \ "$OPAQUE_HELPER_BIN" || die "OPAQUE crypto handshake failed" -# ── 6. DPoP wire protocol — the parts Hurl can't drive ────────────────── +# ── 7. DPoP wire protocol — the parts Hurl can't drive ────────────────── # Each proof carries a fresh jti, current iat, htm/htu matching the # exact request, an ES256 signature, and a threaded nonce — none of # which a declarative .hurl template can compute. See