Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 17 Packages Wiki Activity Actions 1
open-standards/joule-code/joule-code-rs/crates/joule-moq-relay/tests/live_control.rs
Transaction Science d28a9dd719 joule-moq-relay: live_control's tests wait for the summaries they read, each wait on its own bound 
a_rooms_ledger_names_the_grant_of_that_room failed once at a host load of about 130 and passed
three runs again. It waited on one thing and read another: it waited until the relay held no
open session (stats().sessions_open == 0) and then read the ended sessions' summaries and the
room records made of them (hop_record). A session leaves the open sessions as its end begins
(session::cleanup removes it first) and its summary is kept only after its scheduler has
stopped and its counts are closed, which on a loaded host comes later; so the record could be
made from fewer summaries than sessions. The same wait stood before five reads of summaries in
the file (the two rooms' ledgers, the withdrawn grant, the metered grants, the revoked grant).

Each now calls ended(relay, n): every session closed, then the summaries of the n sessions the
relay admitted (one refused at its SETUP has none), each wait on its own STEP bound, and the
count held to n. A three-second delay planted in cleanup before keep_summary fails the old
waits and passes the new; a grant of another room planted in a room's ledger (Tally::grant_in
without its filter) still fails the test.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-11-12 03:12:14 -05:00

1683 lines
103 KiB
Rust
Raw Permalink Blame History

//! Grants that change while the relay runs, and a window closed with its sessions still open
//! (`src/control.rs`), on the operator's surface:
//!
//! - **A withdrawn grant ends its sessions alone.** Two rooms, each a publisher and a
//! subscriber under a token of its own; the first room's token is withdrawn on the control
//! socket: its two sessions are closed `UNAUTHORIZED`, `grant_revoked`, while the other room
//! goes on delivering objects with no GOAWAY. A token added at run time is admitted, the
//! withdrawn one refused, and a metered grant added at run time names the grant in the
//! ledger of a session already open.
//! - **A token a request presented** ends its session the same way when it is withdrawn, and
//! a change that leaves it alone leaves that session alone.
//! - **A metered grant withdrawn** ends the sessions under it, with no rule naming its token,
//! and a `set` that maps the token to another grant moves its session there.
//! - **A change that narrows a rule** ends the sessions holding what the rule no longer covers,
//! a subscription moved off its prefix or a publication whose right went, and none other: a
//! session that loses a right it does not use goes on.
//! - **The operator's surface is local.** A `listen` address off loopback refuses the relay's
//! start; the socket is made with mode 0600; a peer of another user, or off loopback, is
//! refused; a connection from a non-loopback address of this host to the loopback listener
//! is answered with a refusal; a request of another protocol is answered and ends its
//! connection, its body unread.
//! - **A `set` states what it replaces**: one that leaves out `required` or `metered` does not
//! read and changes nothing, and a reload leaves `required` as it stands.
//! - **An authority's revocation list** (WAI live-rooms §5): a grant revoked by its digest or
//! by its viewer's tag is refused at its SETUP, `grant_revoked`, before its proof; a grant
//! revoked while its session is open ends that session alone; and an authority's grant is
//! metered from its own ceilings with no metered grant named. A grant whose funds ceiling and
//! currency break live-rooms §4.3's rule is refused at its SETUP, `grant_invalid`.
//! - **A revocation refuses every use** (live-rooms §5): it ends a session that proved nothing,
//! on a relay that requires no authorization; one a rule also admits, and one whose request
//! presented the grant, whether or not a rule names it; a request that presents a revoked
//! grant is refused, on a relay that requires no authorization too, its session standing and
//! no later change ending it for that refused use; and the list holds once its authority is
//! dropped. The records a session's requests may present are bounded.
//! - **An authority** added while the relay runs is honoured from then, and one dropped ends the
//! sessions that proved its grants and no other authority's.
//! - **A session's grant in a room**: one that holds grants for two rooms is in each room's
//! ledger under that room's grant, and in a third room's under neither, whether an authority
//! signed them or a rule with `proof` and a metered grant name them.
//! - **A revocation stays** (live-rooms §5): through a reload, a `set` that does not repeat it,
//! and a restart, which reads the file the relay kept it in.
//! - **A flush inside a group** (WAI live-metering §3.8): the window is cut after a group's
//! first subgroup and before its second; the second, which arrives after the cut, is counted
//! in the window its group began in, the next group in the next; the record the control
//! socket signs over the first window, with every session still open, passes
//! `wai/live-conformance/live_metering_verify.py`'s reading of a record and its ledger; and
//! each session's windows sum to its whole summary.
//! - **A first flush**: a relay never flushed holds no second copy of its sessions' counts, and
//! its first flush puts everything counted so far in window 0, the ended sessions' among it.
//! - **An interval** seals with nobody asking, and a relay told to hold a ceiling without one does
//! not start.
//! - **A window with a DVR store** (§3.9): a group cut by a flush is stated where its in entries
//! allow, sealed before or after the store wrote it, and every record through the verifier.
//! - **SIGHUP** reads the grants file again: a token it drops ends its session, one it adds
//! is admitted.
// The control socket and SIGHUP, which every test here goes through, are Unix's. The relay off
// Unix serves `control.listen` alone.
#![cfg(unix)]
mod common;
use common::*;
use ed25519_dalek::{Signer, SigningKey};
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Parameter, Parameters, RequestErrorCode, SessionError, Token};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::config::{Authority, DvrConfig, DvrWindow, Grant, MeteredGrant, Revocation};
use joule_moq_relay::control::{admits, own_uid, Peer};
use joule_moq_relay::metering::HopRecord;
use joule_moq_relay::summary::SessionSummary;
use joule_moq_relay::transport::TransportError;
use joule_moq_relay::{RelayConfig, RelayServer};
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;
fn now_us() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
}
fn sha_hex(b: &[u8]) -> String {
hex::encode(joule_moq_relay::auth::sha256(b))
}
/// SIGHUP goes to the whole process, so to every relay here that names a grants file: the tests
/// that name one take turns.
static GRANTS_FILE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
/// A socket path of this test's own, short enough for every host's limit.
fn socket_path(what: &str) -> PathBuf {
std::env::temp_dir().join(format!("wai-relay-ctl-{}-{what}.sock", std::process::id()))
}
/// One request on the control socket at `path`, and its answer.
async fn ask(path: &Path, req: Value) -> Value {
timeout(STEP * 2, async {
let s = tokio::net::UnixStream::connect(path).await.expect("the control socket");
let (r, mut w) = s.into_split();
w.write_all(format!("{req}\n").as_bytes()).await.unwrap();
let mut line = String::new();
BufReader::new(r).read_line(&mut line).await.unwrap();
serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("an answer: {e}: {line:?}"))
})
.await
.expect("an answer in time")
}
fn rule(prefix: &[&str], token: &[u8], publish: bool, subscribe: bool) -> Grant {
Grant { prefix: prefix.iter().map(|s| s.to_string()).collect(), token_sha256: sha_hex(token), publish, subscribe, proof: false, unlimited: false, seek: false }
}
fn bearer(token: &[u8]) -> ClientOptions {
ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: token.to_vec() }], ..ClientOptions::default() }
}
/// The session `c` is closed by the relay with `code`; its reason.
async fn closed_with(c: &Client, code: SessionError, what: &str) -> String {
match timeout(STEP, c.conn.closed()).await.unwrap_or_else(|_| panic!("{what}: not closed in time")) {
TransportError::Closed { code: got, reason, .. } => {
assert_eq!(got, code.0, "{what}: {reason}");
reason
}
other => panic!("{what}: {other:?}"),
}
}
/// Every session closed and the summaries of `n` kept, each wait on its own bound. A session
/// leaves the relay's open sessions as its end begins, and its summary is kept once its scheduler
/// has stopped and its counts are closed (`session::cleanup`), which on a loaded host comes
/// later: so a test that reads the ended sessions' summaries, or a record made of them, waits for
/// the summaries themselves. `n` is every session the relay admitted (one refused at its SETUP
/// has none), and no more are kept.
async fn ended(relay: &joule_moq_relay::Relay, n: usize) {
assert!(until(STEP, || relay.stats().sessions_open == 0).await, "every session closed: {:?}", relay.stats());
assert!(until(STEP, || relay.summaries().len() >= n).await, "{n} sessions' summaries kept: {} so far", relay.summaries().len());
assert_eq!(relay.summaries().len(), n, "a summary of each session admitted, and no other");
}
/// The media streams of `alias` read whole from `got`, by Group ID.
fn groups_of(got: &[Got], alias: u64) -> Vec<u64> {
let mut g: Vec<u64> = got.iter().filter(|x| x.header.track_alias == alias && x.reset.is_none() && !x.objects.is_empty()).map(|x| x.header.group_id).collect();
g.sort();
g.dedup();
g
}
// ── A withdrawn grant ends its sessions alone ───────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn withdrawing_one_grant_ends_only_its_sessions() {
timeout(Duration::from_secs(180), withdraw_run()).await.expect("the test finishes in time");
}
async fn withdraw_run() {
let (ta_tok, tb_tok, tc_tok) = (b"room-a-token".to_vec(), b"room-b-token".to_vec(), b"room-b-late-token".to_vec());
let sock = socket_path("withdraw");
let server = relay(|c| {
c.auth.required = true;
c.auth.grants = vec![rule(&["roomA"], &ta_tok, true, true), rule(&["roomB"], &tb_tok, true, true)];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let ta = Tracks::new(ns(&["roomA", "p"]), "cam", 10, 4, 128);
let tb = Tracks::new(ns(&["roomB", "p"]), "cam", 10, 4, 128);
let (key_a, key_b) = (SigningKey::from_bytes(&[0xa1; 32]), SigningKey::from_bytes(&[0xb2; 32]));
let (pa, _ea) = connect(&server, Via::Quic, bearer(&ta_tok)).await;
let (pb, _eb) = connect(&server, Via::Quic, bearer(&tb_tok)).await;
let _pa = publish_tracks(&pa, &ta, None).await;
let _pb = publish_tracks(&pb, &tb, None).await;
let (sa, _esa) = connect(&server, Via::WebSocket, bearer(&ta_tok)).await;
let (sb, _esb) = connect(&server, Via::Quic, bearer(&tb_tok)).await;
let va = subscribe_both(&sa, &ta, Parameters::new()).await;
let vb = subscribe_both(&sb, &tb, Parameters::new()).await;
publish_group(&pa, &key_a, &ta, 0, &payload("a", 0, 400)).await;
publish_group(&pb, &key_b, &tb, 0, &payload("b", 0, 400)).await;
let got = collect_concurrent(&sa, STEP, Duration::from_millis(300), |g| groups_of(g, va.media_alias) == [0]).await;
assert_eq!(groups_of(&got, va.media_alias), vec![0], "room A delivers before the change");
let got = collect_concurrent(&sb, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [0]).await;
assert_eq!(groups_of(&got, vb.media_alias), vec![0], "room B delivers before the change");
// The grants in force, as the socket shows them.
let shown = ask(&sock, json!({"op": "show"})).await;
assert_eq!(shown["ok"], true, "{shown}");
assert_eq!(shown["auth"]["grants"].as_array().unwrap().len(), 2, "{shown}");
// Room A's token is withdrawn: its two sessions end, with the reason, and only they.
let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(&ta_tok)})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(2)), "{a}");
for (c, what) in [(&pa, "room A's publisher"), (&sa, "room A's subscriber")] {
let why = closed_with(c, SessionError::UNAUTHORIZED, what).await;
assert!(why.starts_with("grant_revoked"), "{what}: {why}");
}
assert!(until(STEP, || relay.stats().sessions_open == 2).await, "room B's two sessions stand");
// A metered grant added now names the grant of room B's sessions, already open.
let grant_b = "b0".repeat(32);
let m = ask(&sock, json!({"op": "meter", "grant": {"token_sha256": sha_hex(&tb_tok), "grant": grant_b, "joules_ceiling_micro": 7_000_000u64, "funds_ceiling": 900, "currency": "USD"}})).await;
assert_eq!((m["ok"].as_bool(), m["ended"].as_u64()), (Some(true), Some(0)), "{m}");
// Room B goes on delivering, with no GOAWAY.
for g in 1..=4 {
publish_group(&pb, &key_b, &tb, g, &payload("b", g, 400)).await;
}
let got = collect_concurrent(&sb, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [1, 2, 3, 4]).await;
assert_eq!(groups_of(&got, vb.media_alias), vec![1, 2, 3, 4], "room B delivers every group after the change");
assert!(pb.goaway().borrow().is_none() && sb.goaway().borrow().is_none(), "no GOAWAY to room B");
// The withdrawn token is refused; a token added at run time is admitted.
let (x, _ex) = connect(&server, Via::Quic, bearer(&ta_tok)).await;
let refused = x.subscribe(ta.ns.clone(), ta.media.clone(), Parameters::new()).await.expect_err("a withdrawn token");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
let (late, _el) = connect(&server, Via::Quic, bearer(&tc_tok)).await;
let refused = late.subscribe(tb.ns.clone(), tb.media.clone(), Parameters::new()).await.expect_err("a token not yet granted");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
let g = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["roomB"], "token_sha256": sha_hex(&tc_tok), "subscribe": true}})).await;
assert_eq!(g["ok"], true, "{g}");
let vl = subscribe_both(&late, &tb, Parameters::new()).await;
publish_group(&pb, &key_b, &tb, 5, &payload("b", 5, 400)).await;
let got = collect_concurrent(&late, STEP, Duration::from_millis(300), |g| groups_of(g, vl.media_alias).contains(&5)).await;
assert!(groups_of(&got, vl.media_alias).contains(&5), "the token granted at run time is served");
// A request that does not read, and a change that does not check, change nothing.
let bad = ask(&sock, json!({"op": "grant", "grant": {"prefix": [], "token_sha256": "zz"}})).await;
assert_eq!(bad["ok"], false, "{bad}");
let bad = ask(&sock, json!({"op": "shutdown"})).await;
assert_eq!(bad["ok"], false, "{bad}");
assert_eq!(relay.auth_config().grants.len(), 2, "room B's rule and the late one");
for c in [&pb, &sb, &x, &late] {
c.close(0, "done");
}
// Room A's two sessions, room B's two, the withdrawn token's and the late one's.
ended(&relay, 6).await;
let summaries: Vec<SessionSummary> = relay.summaries().into_iter().map(|s| s.summary).collect();
let metered: Vec<_> = summaries.iter().filter_map(|s| s.tally.room_grant.as_ref()).collect();
assert_eq!(metered.len(), 2, "room B's publisher and subscriber ran under the grant metered at run time: {metered:?}");
assert!(metered.iter().all(|g| g.grant == grant_b && g.joules_ceiling_micro == 7_000_000 && g.funds_ceiling == 900 && g.currency == "USD"));
server.shutdown().await;
assert!(!sock.exists(), "the socket is removed at shutdown");
}
// ── The operator's surface is local ─────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_control_request_from_outside_is_refused() {
timeout(Duration::from_secs(120), outside_run()).await.expect("the test finishes in time");
}
async fn outside_run() {
// Off loopback, the relay does not start.
for a in ["0.0.0.0:0", "[::]:0", "192.0.2.7:4490"] {
let mut c = RelayConfig::loopback();
c.control.listen = Some(a.parse().unwrap());
let e = RelayServer::start(c).await.err().unwrap_or_else(|| panic!("a control listener on {a} was accepted"));
assert!(e.contains("loopback"), "{a}: {e}");
}
// A peer is served only on the relay's own user's socket, or from loopback.
let me = own_uid();
assert!(admits(&Peer::Unix { uid: me }, me));
assert!(!admits(&Peer::Unix { uid: me.wrapping_add(1).max(1) }, me), "another user's peer");
assert!(admits(&Peer::Tcp("127.0.0.1:5000".parse().unwrap()), me) && admits(&Peer::Tcp("[::1]:5000".parse().unwrap()), me));
for a in ["192.0.2.7:5000", "10.0.0.1:5000", "[2001:db8::7]:5000"] {
assert!(!admits(&Peer::Tcp(a.parse().unwrap()), me), "{a}");
}
let sock = socket_path("outside");
let server = relay(|c| {
c.control.socket = Some(sock.clone());
c.control.listen = Some("127.0.0.1:0".parse().unwrap());
})
.await;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mode = std::fs::metadata(&sock).expect("the socket").permissions().mode() & 0o777;
assert_eq!(mode, 0o600, "the socket is its owner's alone");
}
// From loopback, served.
let addr = server.control_addr.expect("the loopback listener");
let answer = |s: tokio::net::TcpStream| async move {
let (r, mut w) = s.into_split();
let _ = w.write_all(b"{\"op\":\"show\"}\n").await;
let mut line = String::new();
let _ = timeout(STEP, BufReader::new(r).read_line(&mut line)).await;
serde_json::from_str::<Value>(&line).unwrap_or(Value::Null)
};
let a = answer(tokio::net::TcpStream::connect(addr).await.unwrap()).await;
assert_eq!(a["ok"], true, "{a}");
// A request of another protocol is answered and ends its connection, its body unread.
let body = json!({"op": "grant", "grant": {"prefix": ["x"], "token_sha256": "ab".repeat(32), "subscribe": true}}).to_string();
let mut http = tokio::net::TcpStream::connect(addr).await.unwrap();
http.write_all(format!("POST / HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: text/plain\r\nContent-Length: {}\r\n\r\n{body}\n", body.len() + 1).as_bytes()).await.unwrap();
let mut said = String::new();
let _ = timeout(STEP, http.read_to_string(&mut said)).await.expect("the connection ends");
assert!(server.relay().auth_config().grants.is_empty(), "a body after a line of another protocol was obeyed: {said}");
// From an address of this host that is not loopback, to the loopback listener: refused.
// Where the host has no such address, or will not route it to loopback, this part is
// said to be skipped.
let outward = std::net::UdpSocket::bind("0.0.0.0:0").and_then(|u| u.connect("192.0.2.1:9").and_then(|_| u.local_addr())).ok().map(|a| a.ip()).filter(|ip| !ip.is_loopback() && !ip.is_unspecified());
match outward {
Some(ip) => {
let s = tokio::net::TcpSocket::new_v4().unwrap();
match s.bind((ip, 0).into()) {
Ok(()) => match timeout(STEP, s.connect(addr)).await {
Ok(Ok(stream)) => {
assert_eq!(stream.local_addr().unwrap().ip(), ip);
let a = answer(stream).await;
assert_eq!(a["ok"], false, "a peer at {ip} was served: {a}");
assert!(a["error"].as_str().unwrap_or("").contains("refused"), "{a}");
}
other => eprintln!("SKIPPED: a connection from {ip} to {addr}: {other:?}"),
},
Err(e) => eprintln!("SKIPPED: binding {ip}: {e}"),
}
}
None => eprintln!("SKIPPED: this host has no address off loopback"),
}
server.shutdown().await;
}
/// As root, a peer of another user in earnest: it drops to that user, connects to the socket and
/// asks, and the relay refuses it by what the kernel states of it, though the socket's mode
/// would let it in. No other user can be two users on one host, so elsewhere this says it was
/// skipped, and `src/control.rs`'s test of the socket holds the rule.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_peer_of_another_user_is_refused_on_the_socket() {
timeout(Duration::from_secs(60), other_user_run()).await.expect("the test finishes in time");
}
async fn other_user_run() {
if own_uid() != 0 {
eprintln!("SKIPPED: only root connects as another user");
return;
}
use std::os::unix::fs::PermissionsExt;
let sock = socket_path("other-user");
let server = relay(|c| c.control.socket = Some(sock.clone())).await;
std::fs::set_permissions(&sock, std::fs::Permissions::from_mode(0o666)).unwrap();
let req = format!("{}\n", json!({"op": "grant", "grant": {"prefix": ["x"], "token_sha256": "ab".repeat(32), "subscribe": true}}));
// SAFETY: an all-zero sockaddr_un is one with an empty path.
let mut addr: libc::sockaddr_un = unsafe { std::mem::zeroed() };
addr.sun_family = libc::AF_UNIX as _;
for (to, from) in addr.sun_path.iter_mut().zip(sock.as_os_str().as_encoded_bytes()) {
*to = *from as libc::c_char;
}
let status = tokio::task::spawn_blocking(move || {
// SAFETY: the child of a threaded process calls only async-signal-safe functions, on what
// was made before the fork, and leaves by _exit.
unsafe {
let pid = libc::fork();
assert!(pid >= 0, "fork");
if pid == 0 {
let mut answer = [0u8; 256];
let s = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0);
let connected = libc::setuid(65534) == 0 && s >= 0 && libc::connect(s, (&raw const addr).cast(), std::mem::size_of::<libc::sockaddr_un>() as libc::socklen_t) == 0;
// A refusal reads no request, so this write may find the socket closed: the answer
// is there to read all the same.
let n = if connected {
libc::write(s, req.as_ptr().cast(), req.len());
libc::read(s, answer.as_mut_ptr().cast(), answer.len())
} else {
-1
};
let refused = n > 0 && answer[..n as usize].windows(7).any(|w| w == b"refused");
libc::_exit(match (n > 0, refused) {
(true, true) => 0,
(true, false) => 1,
_ => 2,
});
}
let mut status = 0;
assert_eq!(libc::waitpid(pid, &mut status, 0), pid);
status
}
})
.await
.unwrap();
assert!(libc::WIFEXITED(status), "the peer ended by signal: {status}");
assert_eq!(libc::WEXITSTATUS(status), 0, "0 refused, 1 served, 2 no answer");
assert!(server.relay().auth_config().grants.is_empty(), "its request was obeyed");
server.shutdown().await;
}
// ── A token a request presented ─────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_token_a_request_presented_ends_its_session_when_withdrawn() {
timeout(Duration::from_secs(120), presented_run()).await.expect("the test finishes in time");
}
async fn presented_run() {
let (publisher_tok, request_tok, other_tok) = (b"publisher".to_vec(), b"in-a-request".to_vec(), b"another".to_vec());
let sock = socket_path("presented");
let server = relay(|c| {
c.auth.required = true;
c.auth.grants = vec![rule(&["room"], &publisher_tok, true, false), rule(&["room"], &request_tok, false, true)];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let t = Tracks::new(ns(&["room", "p"]), "cam", 10, 4, 128);
let key = SigningKey::from_bytes(&[0xd4; 32]);
let (p, _ep) = connect(&server, Via::Quic, bearer(&publisher_tok)).await;
let _pp = publish_tracks(&p, &t, None).await;
// No token in its SETUP: its subscriptions stand on the token each request carries.
let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: request_tok.clone() }));
let v = subscribe_both_with(&s, &t, with_token.clone(), with_token).await;
publish_group(&p, &key, &t, 0, &payload("r", 0, 300)).await;
let got = collect_concurrent(&s, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
assert_eq!(groups_of(&got, v.media_alias), vec![0], "served on the token of its requests");
// A change that leaves its token alone leaves it alone.
let a = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["elsewhere"], "token_sha256": sha_hex(&other_tok), "subscribe": true}})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(0)), "{a}");
publish_group(&p, &key, &t, 1, &payload("r", 1, 300)).await;
let got = collect_concurrent(&s, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [1]).await;
assert_eq!(groups_of(&got, v.media_alias), vec![1], "it goes on");
// Its token withdrawn, it ends, with the reason; the publisher, under another, does not.
let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(&request_tok)})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
let why = closed_with(&s, SessionError::UNAUTHORIZED, "the session of a request's token").await;
assert!(why.starts_with("grant_revoked") && why.contains("a request presented"), "{why}");
assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the publisher stands");
p.close(0, "done");
server.shutdown().await;
}
// ── A change that narrows a rule ────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_change_that_narrows_a_rule_ends_what_it_no_longer_covers() {
timeout(Duration::from_secs(180), narrow_run()).await.expect("the test finishes in time");
}
async fn narrow_run() {
let (publisher, moved, kept, unused, unpublished) = (b"publisher".as_slice(), b"moved-away".as_slice(), b"kept".as_slice(), b"loses-a-right-unused".as_slice(), b"loses-publish".as_slice());
let waiting = b"waits-for-its-track".as_slice();
let sock = socket_path("narrow");
let server = relay(|c| {
c.auth.required = true;
c.auth.grants = vec![rule(&["roomA", "p"], publisher, true, false), rule(&["roomC"], publisher, true, false), rule(&["roomA"], moved, false, true), rule(&["roomA"], kept, false, true), rule(&["roomA"], unused, true, true), rule(&["roomB"], unpublished, true, false), rule(&["roomC"], waiting, false, true)];
c.control.socket = Some(sock.clone());
c.limits.max_rendezvous_ms = 30_000;
})
.await;
let relay = server.relay().clone();
let ta = Tracks::new(ns(&["roomA", "p"]), "cam", 10, 4, 128);
let tb = Tracks::new(ns(&["roomB", "p"]), "cam", 20, 5, 128);
let key = SigningKey::from_bytes(&[0xe5; 32]);
let (p, _ep) = connect(&server, Via::Quic, bearer(publisher)).await;
let _pp = publish_tracks(&p, &ta, None).await;
let (q, _eq) = connect(&server, Via::Quic, bearer(unpublished)).await;
let _pq = publish_tracks(&q, &tb, None).await;
let (s_moved, _e1) = connect(&server, Via::Quic, bearer(moved)).await;
let (s_kept, _e2) = connect(&server, Via::Quic, bearer(kept)).await;
let (s_unused, _e3) = connect(&server, Via::Quic, bearer(unused)).await;
let v_moved = subscribe_both(&s_moved, &ta, Parameters::new()).await;
let v_kept = subscribe_both(&s_kept, &ta, Parameters::new()).await;
let v_unused = subscribe_both(&s_unused, &ta, Parameters::new()).await;
publish_group(&p, &key, &ta, 0, &payload("n", 0, 300)).await;
for (c, v) in [(&s_moved, &v_moved), (&s_kept, &v_kept), (&s_unused, &v_unused)] {
let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
assert_eq!(groups_of(&got, v.media_alias), vec![0], "delivered before the change");
}
// A subscription admitted now, waiting for a track nobody publishes yet, holds nothing.
let tc = Tracks::new(ns(&["roomC", "p"]), "cam", 30, 6, 128);
let s_waiting = std::sync::Arc::new(connect(&server, Via::Quic, bearer(waiting)).await);
let pending = {
let (c, t) = (s_waiting.clone(), tc.clone());
tokio::spawn(async move { c.0.subscribe(t.ns.clone(), t.media.clone(), Parameters::new().with(Parameter::RendezvousTimeout(30_000))).await.map(|_| ()) })
};
tokio::time::sleep(Duration::from_secs(1)).await;
// Every token stays named. One moves to another prefix, one loses the right to publish that
// its session uses, one the right its session does not use, one the prefix its session is
// still waiting on.
let next = [rule(&["roomA", "p"], publisher, true, false), rule(&["roomC"], publisher, true, false), rule(&["roomZ"], moved, false, true), rule(&["roomA"], kept, false, true), rule(&["roomA"], unused, false, true), rule(&["roomB"], unpublished, false, true), rule(&["roomZ"], waiting, false, true)];
let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": next, "authorities": []}, "metered": []})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(2)), "{a}");
for (c, what) in [(&s_moved, "a subscription its rule no longer covers"), (&q, "a publication whose right went")] {
let why = closed_with(c, SessionError::UNAUTHORIZED, what).await;
assert!(why.starts_with("grant_revoked") && why.contains("no longer cover"), "{what}: {why}");
}
assert!(until(STEP, || relay.stats().sessions_open == 4).await, "the others stand, the waiting session among them");
// Its track appears: the session that waited meets the grants in force as it takes its place.
// On a host so loaded that the request came after the change, it was refused as it arrived.
if pending.is_finished() {
eprintln!("SKIPPED: the waiting request reached the relay after the change: {:?}", pending.await);
s_waiting.0.close(0, "done");
} else {
let _pc = publish_tracks(&p, &tc, None).await;
let why = closed_with(&s_waiting.0, SessionError::UNAUTHORIZED, "a subscription placed after the change").await;
assert!(why.starts_with("grant_revoked") && why.contains("no longer cover"), "{why}");
let _ = timeout(STEP, pending).await.expect("the request ends");
}
publish_group(&p, &key, &ta, 1, &payload("n", 1, 300)).await;
for (c, v) in [(&s_kept, &v_kept), (&s_unused, &v_unused)] {
let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [1]).await;
assert_eq!(groups_of(&got, v.media_alias), vec![1], "the rest goes on delivering");
}
for c in [&p, &s_kept, &s_unused] {
c.close(0, "done");
}
server.shutdown().await;
}
// ── A set states what it replaces ───────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_set_states_what_it_replaces() {
let _turn = GRANTS_FILE.lock().await;
timeout(Duration::from_secs(120), set_run()).await.expect("the test finishes in time");
}
async fn set_run() {
let tok = b"a-token".to_vec();
let sock = socket_path("set");
let file = std::env::temp_dir().join(format!("wai-relay-set-{}.toml", std::process::id()));
std::fs::write(&file, grants_file(&[])).unwrap();
let server = relay(|c| {
c.control.socket = Some(sock.clone());
c.control.grants_path = Some(file.clone());
})
.await;
let relay = server.relay().clone();
let t = Tracks::new(ns(&["live", "s"]), "cam", 10, 4, 128);
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _pp = publish_tracks(&p, &t, None).await;
let anonymous = || async {
let (c, e) = connect(&server, Via::Quic, ClientOptions::default()).await;
let r = c.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.map(|_| ());
c.close(0, "done");
drop(e);
r
};
anonymous().await.expect("open, as the configuration has it");
assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the publisher alone");
// Stated, it closes the relay; the publisher, with no token, holds what it no longer may.
let metered = json!({"token_sha256": sha_hex(&tok), "grant": "c0".repeat(32)});
let rules = [rule(&["live"], &tok, true, true)];
let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": rules, "authorities": []}, "metered": [metered]})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
let (p2, _e2) = connect(&server, Via::Quic, bearer(&tok)).await;
let _pp2 = publish_tracks(&p2, &t, None).await;
assert!(anonymous().await.is_err(), "required, as the set stated");
// Left out, each changes nothing: the relay stays closed, and the metered grant stands.
let a = ask(&sock, json!({"op": "set", "auth": {"grants": rules, "authorities": []}, "metered": []})).await;
assert_eq!(a["ok"], false, "a set with no required: {a}");
let a = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": rules, "authorities": []}})).await;
assert_eq!(a["ok"], false, "a set with no metered: {a}");
assert!(relay.auth_config().required && relay.metered_grants().len() == 1);
assert!(anonymous().await.is_err(), "still required");
// A reload takes the file's lists, and leaves whether authorization is required as it stands.
std::fs::write(&file, grants_file(&[(tok.as_slice(), true, true)])).unwrap();
let a = ask(&sock, json!({"op": "reload"})).await;
assert_eq!(a["ok"], true, "{a}");
assert!(relay.auth_config().required, "the configuration's false is not taken back by a reload");
assert!(anonymous().await.is_err(), "still required after the reload");
p2.close(0, "done");
let _ = std::fs::remove_file(&file);
server.shutdown().await;
}
// ── A metered grant withdrawn ───────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn withdrawing_a_metered_grant_ends_the_sessions_under_it() {
timeout(Duration::from_secs(120), metered_run()).await.expect("the test finishes in time");
}
async fn metered_run() {
let (one, two, three) = (b"metered-one".as_slice(), b"metered-two".as_slice(), b"metered-three".as_slice());
let grant = |n: u8| hex::encode([n; 32]);
let metered = |t: &[u8], n: u8| MeteredGrant { token_sha256: sha_hex(t), grant: grant(n), joules_ceiling_micro: u64::MAX, funds_ceiling: u64::MAX, currency: String::new() };
let sock = socket_path("metered");
let server = relay(|c| {
c.metering.grants = vec![metered(one, 0xa1), metered(two, 0xa2), metered(three, 0xa3)];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let (s1, _e1) = connect(&server, Via::Quic, bearer(one)).await;
let (s2, _e2) = connect(&server, Via::Quic, bearer(two)).await;
let (s3, _e3) = connect(&server, Via::Quic, bearer(three)).await;
assert!(until(STEP, || relay.stats().sessions_open == 3).await);
// No rule names the token: its metered grant is all that does, and the session ends with it.
let a = ask(&sock, json!({"op": "withdraw", "token_sha256": sha_hex(one)})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
let why = closed_with(&s1, SessionError::UNAUTHORIZED, "the session under the withdrawn metered grant").await;
assert!(why.starts_with("grant_revoked") && why.contains("its SETUP presented"), "{why}");
// A set that leaves a token out ends its session, and one that maps a token to another grant
// moves its session there.
let a = ask(&sock, json!({"op": "set", "auth": {"required": false, "grants": [], "authorities": []}, "metered": [metered(three, 0xb3)]})).await;
assert_eq!((a["ok"].as_bool(), a["ended"].as_u64()), (Some(true), Some(1)), "{a}");
let why = closed_with(&s2, SessionError::UNAUTHORIZED, "the session of the token the set left out").await;
assert!(why.starts_with("grant_revoked"), "{why}");
assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the session whose token stays mapped goes on");
assert_eq!(relay.open_tallies()[0].1.room_grant.as_ref().map(|g| g.grant.clone()), Some(grant(0xb3)));
s3.close(0, "done");
ended(&relay, 3).await;
let mut named: Vec<String> = relay.summaries().into_iter().filter_map(|s| s.summary.tally.room_grant.map(|g| g.grant)).collect();
named.sort();
assert_eq!(named, vec![grant(0xa1), grant(0xa2), grant(0xb3)], "each session's summary names what it ran under");
server.shutdown().await;
}
// ── An authority's revocation list ──────────────────────────────────────────
fn now_s() -> u64 {
now_us() / 1_000_000
}
/// A grant laid out as live-rooms §4.1 lays it out, signed by `authority`: `publish` the media
/// bits, a policy that permits the chat kind when `events`, its `viewer`, its ceilings.
#[allow(clippy::too_many_arguments)]
fn room_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, events: bool, viewer: [u8; 16], joules: u64, funds: u64, currency: &[u8; 3]) -> Vec<u8> {
let now = now_s();
let mut g = b"WRG1\x01".to_vec();
g.extend_from_slice(&room);
g.extend_from_slice(&principal.verifying_key().to_bytes());
g.extend_from_slice(&[0; 40]);
g.extend_from_slice(&(now - 60).to_le_bytes());
g.extend_from_slice(&(now + 600).to_le_bytes());
g.extend_from_slice(&[0, publish, 1, 0, 0, 0]);
g.extend_from_slice(&[0; 16]); // once
g.extend_from_slice(&viewer);
g.extend_from_slice(b"WEP1\x01");
let mut kinds = [0u8; 32];
if events {
kinds[0] = 1 << 3;
}
g.extend_from_slice(&kinds);
g.extend_from_slice(&[0; 28]);
g.extend_from_slice(&joules.to_le_bytes());
g.extend_from_slice(&funds.to_le_bytes());
g.extend_from_slice(currency);
g.extend_from_slice(&[0, 0, 0, 0]);
let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
m.extend_from_slice(&g);
g.extend_from_slice(&authority.sign(&m).to_bytes());
g
}
fn digest_hex(g: &[u8]) -> String {
hex::encode(joule_moq::grant_proof::grant_digest_and_principal(g).unwrap().0)
}
fn viewer_tag_hex(viewer: [u8; 16]) -> String {
let mut c = ring::digest::Context::new(&ring::digest::SHA256);
c.update(b"wai:live-viewer\x01");
c.update(&viewer);
hex::encode(c.finish().as_ref())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_authoritys_revoked_grant_is_refused_at_its_proof() {
timeout(Duration::from_secs(180), revocation_run()).await.expect("the test finishes in time");
}
async fn revocation_run() {
let authority = SigningKey::from_bytes(&[0x61; 32]);
let ak = hex::encode(authority.verifying_key().to_bytes());
let room = [0x7e; 32];
let r = hex::encode(room);
let keys: Vec<SigningKey> = (0..5u8).map(|i| SigningKey::from_bytes(&[0x70 + i; 32])).collect();
let none = [0u8; 16];
let alice = room_grant(&authority, room, &keys[0], 0b011, true, none, u64::MAX, u64::MAX, &[0; 3]);
let by_digest = room_grant(&authority, room, &keys[1], 0, false, none, u64::MAX, u64::MAX, &[0; 3]);
let by_viewer = room_grant(&authority, room, &keys[2], 0, false, [0x5c; 16], u64::MAX, u64::MAX, &[0; 3]);
let later = room_grant(&authority, room, &keys[3], 0, false, [0x5d; 16], u64::MAX, u64::MAX, &[0; 3]);
let metered = room_grant(&authority, room, &keys[4], 0, false, [0x5e; 16], 5_000_000, 500, b"EUR");
let sock = socket_path("revoke");
let server = relay(|c| {
c.auth.required = true;
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
c.auth.revocations = vec![Revocation { authority: ak.clone(), revoked: vec![digest_hex(&by_digest), viewer_tag_hex([0x5c; 16])] }];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };
// Revoked by its digest, or by its viewer's tag: refused at SETUP, before its proof.
for (g, k, what) in [(&by_digest, &keys[1], "revoked by its digest"), (&by_viewer, &keys[2], "revoked by its viewer's tag")] {
let (c, _) = connect(&server, Via::WebSocket, with(g, k)).await;
let why = closed_with(&c, SessionError::UNAUTHORIZED, what).await;
assert!(why.contains("grant_revoked"), "{what}: {why}");
}
// Alice publishes; two viewers under grants not revoked subscribe.
let ta = Tracks::new(ns(&[&r, &hex::encode(keys[0].verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
let (a, _) = connect(&server, Via::WebSocket, with(&alice, &keys[0])).await;
let _pa = publish_tracks(&a, &ta, None).await;
let (v3, _) = connect(&server, Via::WebSocket, with(&later, &keys[3])).await;
let (v4, _) = connect(&server, Via::WebSocket, with(&metered, &keys[4])).await;
let s3 = subscribe_both(&v3, &ta, Parameters::new()).await;
let s4 = subscribe_both(&v4, &ta, Parameters::new()).await;
publish_group(&a, &keys[0], &ta, 0, &payload("a", 0, 300)).await;
for (v, s) in [(&v3, &s3), (&v4, &s4)] {
let got = collect_concurrent(v, STEP, Duration::from_millis(300), |g| groups_of(g, s.media_alias) == [0]).await;
assert_eq!(groups_of(&got, s.media_alias), vec![0]);
}
// The authority revokes the third viewer's grant while its session is open: that session
// ends, and the room goes on for the others.
let x = ask(&sock, json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(&later)]})).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&v3, SessionError::UNAUTHORIZED, "a grant revoked while it ran").await;
assert!(why.contains("grant_revoked") && why.contains("live-rooms §5"), "{why}");
publish_group(&a, &keys[0], &ta, 1, &payload("a", 1, 300)).await;
let got = collect_concurrent(&v4, STEP, Duration::from_millis(300), |g| groups_of(g, s4.media_alias) == [1]).await;
assert_eq!(groups_of(&got, s4.media_alias), vec![1], "the room goes on");
let (again, _) = connect(&server, Via::WebSocket, with(&later, &keys[3])).await;
let why = closed_with(&again, SessionError::UNAUTHORIZED, "the revoked grant presented again").await;
assert!(why.contains("grant_revoked"), "{why}");
// The metered viewer's ledger entry is its grant's own: its digest and its ceilings.
v4.close(0, "done");
a.close(0, "done");
// Alice's and the two viewers' (the three refused at SETUP have none).
ended(&relay, 3).await;
let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grant.as_ref().is_some_and(|g| g.grant == digest_hex(&metered))).expect("the viewer's summary names its grant");
let g = s.tally.room_grant.unwrap();
assert_eq!((g.joules_ceiling_micro, g.funds_ceiling, g.currency.as_str()), (5_000_000, 500, "EUR"));
let rec = relay.hop_record(room, now_us() - 600_000_000);
let e = rec.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&metered)).expect("the ledger names the grant");
assert_eq!((e.joules_ceiling, e.funds_ceiling, &e.currency), (5_000_000, 500, b"EUR"));
let ea = rec.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&alice)).expect("alice's grant");
assert_eq!((ea.joules_ceiling, ea.funds_ceiling, ea.currency), (u64::MAX, u64::MAX, [0; 3]), "no ceiling: none");
conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
server.shutdown().await;
}
/// live-rooms §4.3 step 5: a grant whose funds ceiling and currency disagree is `grant_invalid`,
/// whoever signed it, and so never reaches a ledger, whose rule (live-metering §3.6) it would
/// break for every grant of the record.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_grant_whose_funds_and_currency_disagree_is_refused() {
timeout(Duration::from_secs(120), invalid_run()).await.expect("the test finishes in time");
}
async fn invalid_run() {
let authority = SigningKey::from_bytes(&[0x63; 32]);
let ak = hex::encode(authority.verifying_key().to_bytes());
let room = [0x7d; 32];
let viewer = SigningKey::from_bytes(&[0x75; 32]);
for required in [true, false] {
let server = relay(|c| {
c.auth.required = required;
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
})
.await;
let relay = server.relay().clone();
let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };
for (funds, currency, what) in [(500, [0u8; 3], "funds with no currency"), (500, *b"eur", "a currency not in capitals"), (500, [0xff, b'U', b'R'], "a currency not text"), (u64::MAX, *b"EUR", "a currency with no funds")] {
let g = room_grant(&authority, room, &viewer, 0, false, [0; 16], 5_000_000, funds, &currency);
let (c, _) = connect(&server, Via::WebSocket, with(&g, &viewer)).await;
let why = closed_with(&c, SessionError::UNAUTHORIZED, what).await;
assert!(why.contains("grant_invalid"), "{what}: {why}");
}
// A grant that keeps the rule is admitted.
let good = room_grant(&authority, room, &viewer, 0, false, [0; 16], 5_000_000, 500, b"EUR");
let (c, _) = connect(&server, Via::WebSocket, with(&good, &viewer)).await;
assert!(until(STEP, || relay.stats().sessions_open == 1).await, "the session stands");
c.close(0, "done");
assert!(until(STEP, || relay.stats().sessions_open == 0).await);
server.shutdown().await;
}
}
// ── A revocation refuses every use ──────────────────────────────────────────
/// A viewer's grant of `authority` for `room`, with no ceiling.
fn viewer_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey) -> Vec<u8> {
room_grant(authority, room, principal, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3])
}
/// A session on `server` presenting `grants`, each with its proof.
fn proving(server: &RelayServer, grants: &[(&Vec<u8>, &SigningKey)]) -> ClientOptions {
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
ClientOptions { grants: grants.iter().map(|(g, k)| ((*g).clone(), (*k).clone())).collect(), target, ..ClientOptions::default() }
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_revocation_ends_its_session_whatever_admitted_it() {
timeout(Duration::from_secs(240), every_use_run()).await.expect("the test finishes in time");
}
async fn every_use_run() {
let authority = SigningKey::from_bytes(&[0x65; 32]);
let ak = hex::encode(authority.verifying_key().to_bytes());
let room = [0x7b; 32];
let r = hex::encode(room);
let (k_gone, k_stands, k_asked) = (SigningKey::from_bytes(&[0x78; 32]), SigningKey::from_bytes(&[0x79; 32]), SigningKey::from_bytes(&[0x7a; 32]));
let gone = viewer_grant(&authority, room, &k_gone);
let stands = viewer_grant(&authority, room, &k_stands);
let asked = viewer_grant(&authority, room, &k_asked);
let revoke = |g: &[u8]| json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(g)]});
// No authorization required: the relay reads no proof, so the session proved nothing, and the
// list is all that stands against the grant.
let sock = socket_path("use-open");
let server = relay(|c| {
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
c.control.socket = Some(sock.clone());
})
.await;
let open = server.relay().clone();
let t = Tracks::new(ns(&[&r, "p"]), "cam", 10, 4, 128);
let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _pp = publish_tracks(&p, &t, None).await;
let (a, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
let (b, _) = connect(&server, Via::WebSocket, proving(&server, &[(&stands, &k_stands)])).await;
// A session whose requests alone present a grant no rule names: the relay serves it on no
// grant at all, and the revocation reaches it all the same.
let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: asked.clone() }));
let _vs = subscribe_both_with(&s, &t, with_token.clone(), with_token.clone()).await;
assert!(until(STEP, || open.stats().sessions_open == 4).await);
let x = ask(&sock, revoke(&gone)).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&a, SessionError::UNAUTHORIZED, "a session of an open relay that presented the grant").await;
assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
assert!(until(STEP, || open.stats().sessions_open == 3).await, "the other grant's session stands");
let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
let why = closed_with(&again, SessionError::UNAUTHORIZED, "the revoked grant presented again").await;
assert!(why.contains("grant_revoked"), "{why}");
let x = ask(&sock, revoke(&asked)).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&s, SessionError::UNAUTHORIZED, "a session of an open relay whose request presented the grant").await;
assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
// A request that presents it now is refused, as a SETUP is. Its session stands: a request
// without the grant is served, and a later change does not end it for the refused one.
let (late, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
let refused = late.subscribe(t.ns.clone(), t.media.clone(), with_token).await.expect_err("a revoked grant in a request, on an open relay");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()) && refused.contains("grant_revoked"), "{refused}");
let _vl = subscribe_both(&late, &t, Parameters::new()).await;
let x = ask(&sock, json!({"op": "grant", "grant": {"prefix": ["elsewhere"], "token_sha256": sha_hex(b"unrelated"), "subscribe": true}})).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
for c in [&p, &b, &late] {
c.close(0, "done");
}
server.shutdown().await;
// A rule names each grant beside its authority, and goes on naming it after the revocation.
let sock = socket_path("use-rule");
let by_proof = |g: &[u8]| Grant { proof: true, ..rule(&[&r], g, false, true) };
let publisher = b"a-publisher".as_slice();
let server = relay(|c| {
c.auth.required = true;
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
c.auth.grants = vec![by_proof(&gone), by_proof(&stands), rule(&[&r], &asked, false, true), rule(&[&r], publisher, true, false)];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let t = Tracks::new(ns(&[&r, "p"]), "cam", 10, 4, 128);
let key = SigningKey::from_bytes(&[0xd5; 32]);
let (p, _ep) = connect(&server, Via::Quic, bearer(publisher)).await;
let _pp = publish_tracks(&p, &t, None).await;
let (a, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
let (b, _) = connect(&server, Via::WebSocket, proving(&server, &[(&stands, &k_stands)])).await;
let va = subscribe_both(&a, &t, Parameters::new()).await;
let vb = subscribe_both(&b, &t, Parameters::new()).await;
// A session that presents the third grant in its requests alone, under a rule that asks for
// no proof.
let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let with_token = Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: asked.clone() }));
let vs = subscribe_both_with(&s, &t, with_token.clone(), with_token.clone()).await;
publish_group(&p, &key, &t, 0, &payload("u", 0, 300)).await;
for (c, v) in [(&a, &va), (&b, &vb), (&s, &vs)] {
let got = collect_concurrent(c, STEP, Duration::from_millis(300), |g| groups_of(g, v.media_alias) == [0]).await;
assert_eq!(groups_of(&got, v.media_alias), vec![0], "delivered before the revocation");
}
let x = ask(&sock, revoke(&gone)).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&a, SessionError::UNAUTHORIZED, "a session a rule admitted").await;
assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
let why = closed_with(&again, SessionError::UNAUTHORIZED, "the grant presented again under its rule").await;
assert!(why.contains("grant_revoked"), "{why}");
// The grant its requests presented: the session ends, and a request presenting it is refused.
let x = ask(&sock, revoke(&asked)).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&s, SessionError::UNAUTHORIZED, "a session that presented the grant in a request").await;
assert!(why.starts_with("grant_revoked") && why.contains("live-rooms §5"), "{why}");
let (late, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
let refused = late.subscribe(t.ns.clone(), t.media.clone(), with_token).await.expect_err("a revoked grant in a request");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
publish_group(&p, &key, &t, 1, &payload("u", 1, 300)).await;
let got = collect_concurrent(&b, STEP, Duration::from_millis(300), |g| groups_of(g, vb.media_alias) == [1]).await;
assert_eq!(groups_of(&got, vb.media_alias), vec![1], "the room goes on");
late.close(0, "done");
assert!(until(STEP, || relay.stats().sessions_open == 2).await, "the publisher and the grant that stands");
// Its authority dropped, the rule still names the grant, and the list still holds it.
let x = ask(&sock, json!({"op": "drop_authority", "key": ak})).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
let (again, _) = connect(&server, Via::WebSocket, proving(&server, &[(&gone, &k_gone)])).await;
let why = closed_with(&again, SessionError::UNAUTHORIZED, "the grant after its authority went").await;
assert!(why.contains("grant_revoked"), "{why}");
for c in [&p, &b] {
c.close(0, "done");
}
server.shutdown().await;
}
/// A session's requests may present records no rule names up to the relay's bound, each counted
/// once; a request past it is refused, and one that presents none is still answered.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_records_a_sessions_requests_present_are_bounded() {
timeout(Duration::from_secs(120), bound_run()).await.expect("the test finishes in time");
}
async fn bound_run() {
let unauthorized = RequestErrorCode::UNAUTHORIZED.to_string();
let server = relay(|_| {}).await;
let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let record = |n: u8| {
let mut g = b"WRG1\x01".to_vec();
g.resize(joule_moq_relay::relay::PRESENTED_BYTES / 4, n);
g
};
// No such track: what the request is answered is all this test reads.
let answer = async |n: Option<u8>| {
let params = n.map_or_else(Parameters::new, |n| Parameters::new().with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: record(n) })));
s.subscribe(ns(&["room", "p"]), name("none"), params).await.expect_err("nobody publishes it")
};
for n in 0..4 {
let why = answer(Some(n)).await;
assert!(!why.contains(&unauthorized), "within the bound: {why}");
}
let why = answer(Some(4)).await;
assert!(why.contains(&unauthorized) && why.contains("past the grants"), "past it: {why}");
let why = answer(Some(0)).await;
assert!(!why.contains(&unauthorized), "one it holds takes no more room: {why}");
let why = answer(None).await;
assert!(!why.contains(&unauthorized), "a request presenting none: {why}");
s.close(0, "done");
server.shutdown().await;
}
// ── An authority added and dropped ──────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_authority_added_is_honoured_and_one_dropped_ends_its_sessions_alone() {
timeout(Duration::from_secs(180), authorities_run()).await.expect("the test finishes in time");
}
async fn authorities_run() {
let (first, second, added) = (SigningKey::from_bytes(&[0x66; 32]), SigningKey::from_bytes(&[0x67; 32]), SigningKey::from_bytes(&[0x68; 32]));
let key_of = |k: &SigningKey| hex::encode(k.verifying_key().to_bytes());
let (room_1, room_2, room_3) = ([0x71; 32], [0x72; 32], [0x73; 32]);
let (viewer, performer) = (SigningKey::from_bytes(&[0x7d; 32]), SigningKey::from_bytes(&[0x7e; 32]));
let (g1, g2, g3) = (viewer_grant(&first, room_1, &viewer), viewer_grant(&second, room_2, &viewer), viewer_grant(&added, room_3, &viewer));
let g3_publisher = room_grant(&added, room_3, &performer, 0b011, true, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
let sock = socket_path("authorities");
let server = relay(|c| {
c.auth.required = true;
c.auth.authorities = vec![Authority { key: key_of(&first), rooms: vec![] }, Authority { key: key_of(&second), rooms: vec![] }];
c.control.socket = Some(sock.clone());
})
.await;
let relay = server.relay().clone();
let (s1, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g1, &viewer)])).await;
let (s2, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g2, &viewer)])).await;
assert!(until(STEP, || relay.stats().sessions_open == 2).await);
// Not honoured yet: its grant gives nothing.
let t = Tracks::new(ns(&[&hex::encode(room_3), &key_of(&performer), "video"]), "0", 10, 5, 128);
let (early, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3, &viewer)])).await;
let refused = early.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect_err("an authority not honoured yet");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
early.close(0, "done");
assert!(until(STEP, || relay.stats().sessions_open == 2).await);
// Added at run time: its grants are read, proved and served.
let x = ask(&sock, json!({"op": "authority", "authority": {"key": key_of(&added)}})).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(0)), "{x}");
let shown = ask(&sock, json!({"op": "show"})).await;
assert_eq!(shown["auth"]["authorities"].as_array().map(Vec::len), Some(3), "{shown}");
let (p, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3_publisher, &performer)])).await;
let _pp = publish_tracks(&p, &t, None).await;
let (s3, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g3, &viewer)])).await;
let v3 = subscribe_both(&s3, &t, Parameters::new()).await;
publish_group(&p, &performer, &t, 0, &payload("a", 0, 300)).await;
let got = collect_concurrent(&s3, STEP, Duration::from_millis(300), |g| groups_of(g, v3.media_alias) == [0]).await;
assert_eq!(groups_of(&got, v3.media_alias), vec![0], "the authority added at run time is honoured");
// Dropped: the sessions that proved its grants end, though they hold nothing, and only they.
let x = ask(&sock, json!({"op": "drop_authority", "key": key_of(&first)})).await;
assert_eq!((x["ok"].as_bool(), x["ended"].as_u64()), (Some(true), Some(1)), "{x}");
let why = closed_with(&s1, SessionError::UNAUTHORIZED, "the session of the authority dropped").await;
assert!(why.starts_with("grant_revoked") && why.contains("its SETUP presented"), "{why}");
assert!(until(STEP, || relay.stats().sessions_open == 3).await, "the other authorities' sessions stand");
publish_group(&p, &performer, &t, 1, &payload("a", 1, 300)).await;
let got = collect_concurrent(&s3, STEP, Duration::from_millis(300), |g| groups_of(g, v3.media_alias) == [1]).await;
assert_eq!(groups_of(&got, v3.media_alias), vec![1], "another authority's room goes on");
// It is no authority now: dropped again it is not found, and its grant gives nothing.
let x = ask(&sock, json!({"op": "drop_authority", "key": key_of(&first)})).await;
assert_eq!(x["ok"], false, "{x}");
let (late, _) = connect(&server, Via::WebSocket, proving(&server, &[(&g1, &viewer)])).await;
let room_1_track = Tracks::new(ns(&[&hex::encode(room_1), &key_of(&performer), "video"]), "0", 20, 6, 128);
let refused = late.subscribe(room_1_track.ns.clone(), room_1_track.media.clone(), Parameters::new()).await.expect_err("an authority dropped");
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
for c in [&s2, &p, &s3, &late] {
c.close(0, "done");
}
server.shutdown().await;
}
// ── A session's grant in a room ─────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rooms_ledger_names_the_grant_of_that_room() {
timeout(Duration::from_secs(180), two_rooms_run()).await.expect("the test finishes in time");
}
async fn two_rooms_run() {
let t0 = now_us() - 1_000_000;
let authority = SigningKey::from_bytes(&[0x69; 32]);
let ak = hex::encode(authority.verifying_key().to_bytes());
let (room_a, room_b, room_c) = ([0x5a; 32], [0x5b; 32], [0x5c; 32]);
let (viewer, performer, other) = (SigningKey::from_bytes(&[0x6a; 32]), SigningKey::from_bytes(&[0x6b; 32]), SigningKey::from_bytes(&[0x6c; 32]));
let in_a = room_grant(&authority, room_a, &viewer, 0, false, [0; 16], 1_000_000, u64::MAX, &[0; 3]);
let in_b = room_grant(&authority, room_b, &viewer, 0, false, [0; 16], 2_000_000, 300, b"EUR");
let publisher = room_grant(&authority, room_b, &performer, 0b011, true, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
let room_c_token = b"room-c".as_slice();
let server = relay(|c| {
c.auth.required = true;
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
c.auth.grants = vec![rule(&[&hex::encode(room_c)], room_c_token, true, true)];
})
.await;
let relay = server.relay().clone();
let tb = Tracks::new(ns(&[&hex::encode(room_b), &hex::encode(performer.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
let tc = Tracks::new(ns(&[&hex::encode(room_c), &hex::encode(other.verifying_key().to_bytes()), "video"]), "0", 20, 6, 128);
let (p, _) = connect(&server, Via::WebSocket, proving(&server, &[(&publisher, &performer)])).await;
let _pp = publish_tracks(&p, &tb, None).await;
let (q, _eq) = connect(&server, Via::Quic, bearer(room_c_token)).await;
let _pq = publish_tracks(&q, &tc, None).await;
// Grants for rooms A and B, A's first, and a token for room C; it subscribes in B and C.
let holds = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: room_c_token.to_vec() }], ..proving(&server, &[(&in_a, &viewer), (&in_b, &viewer)]) };
let (v, _) = connect(&server, Via::WebSocket, holds).await;
let sb = subscribe_both(&v, &tb, Parameters::new()).await;
let sc = subscribe_both(&v, &tc, Parameters::new()).await;
publish_group(&p, &performer, &tb, 0, &payload("b", 0, 300)).await;
publish_group(&q, &other, &tc, 0, &payload("c", 0, 300)).await;
let got = collect_concurrent(&v, STEP, Duration::from_millis(300), |g| groups_of(g, sb.media_alias) == [0] && groups_of(g, sc.media_alias) == [0]).await;
assert_eq!((groups_of(&got, sb.media_alias), groups_of(&got, sc.media_alias)), (vec![0], vec![0]));
let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
assert!(until(STEP, || forwarded(&relay) == 4).await, "both rooms' objects and receipts counted out");
let grants_of = |rec: &HopRecord| rec.ledger.iter().map(|e| hex::encode(e.grant)).collect::<Vec<_>>();
let holds_to_its_room = |what: &str, b: &HopRecord, c: &HopRecord| {
let names = grants_of(b);
assert!(names.contains(&digest_hex(&in_b)) && names.contains(&digest_hex(&publisher)), "{what}: room B's grants: {names:?}");
assert!(!names.contains(&digest_hex(&in_a)), "{what}: room A's grant in room B's ledger: {names:?}");
let e = b.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&in_b)).expect("the viewer's grant");
assert_eq!((e.joules_ceiling, e.funds_ceiling, &e.currency, e.sessions), (2_000_000, 300, b"EUR", 1), "{what}: its ceilings");
assert!(e.bytes_out > 0 && e.subscriber_us.iter().sum::<u64>() > 0, "{what}: its bytes and its time count against it");
assert_eq!(grants_of(c), vec!["00".repeat(32)], "{what}: no grant of room C");
assert_eq!(c.ledger[0].sessions, 2, "{what}: its publisher and the viewer");
};
// Over a window of the sessions, still open, and again over the sessions ended.
let w = relay.flush_and_seal(Duration::from_millis(200)).await;
let (b, c) = (relay.window_record(room_b, t0, w.index).unwrap(), relay.window_record(room_c, t0, w.index).unwrap());
holds_to_its_room("a window", &b, &c);
conformance_reads(&b.to_bytes(), &b.ledger_bytes(), &room_b, 1);
conformance_reads(&c.to_bytes(), &c.ledger_bytes(), &room_c, 1);
assert!(relay.window_record(room_a, t0, w.index).unwrap().ledger.is_empty(), "nothing of room A was held");
for x in [&p, &q, &v] {
x.close(0, "done");
}
ended(&relay, 3).await;
holds_to_its_room("the sessions ended", &relay.hop_record(room_b, t0), &relay.hop_record(room_c, t0));
let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grants.len() == 2).expect("the viewer's summary names each grant");
assert_eq!(s.tally.room_grants[&hex::encode(room_a)].grant, digest_hex(&in_a));
assert_eq!(s.tally.room_grants[&hex::encode(room_b)].grant, digest_hex(&in_b));
server.shutdown().await;
}
/// A relay that holds no authority and names each grant by a rule with `proof` and a metered
/// grant: a session that proved one grant for each of two rooms runs in each under that room's,
/// the record of each stating its room, and a token that is no record names the rooms its grants
/// do not.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_rooms_ledger_names_the_grant_of_that_room_with_no_authority() {
timeout(Duration::from_secs(180), rooms_by_rule_run()).await.expect("the test finishes in time");
}
async fn rooms_by_rule_run() {
let t0 = now_us() - 1_000_000;
let authority = SigningKey::from_bytes(&[0x6d; 32]);
let (room_a, room_b, room_c) = ([0x5d; 32], [0x5e; 32], [0x5f; 32]);
let (rb, rc) = (hex::encode(room_b), hex::encode(room_c));
let (viewer, performer, other) = (SigningKey::from_bytes(&[0x6e; 32]), SigningKey::from_bytes(&[0x6f; 32]), SigningKey::from_bytes(&[0x70; 32]));
let in_a = viewer_grant(&authority, room_a, &viewer);
let in_b = viewer_grant(&authority, room_b, &viewer);
let (publisher_token, room_c_token) = (b"room-b-publisher".as_slice(), b"room-c".as_slice());
let of_c = "c7".repeat(32);
let by_proof = |room: &[u8; 32], g: &[u8]| Grant { proof: true, ..rule(&[&hex::encode(room)], g, false, true) };
let maps = |t: &[u8], grant: &str, joules: u64| MeteredGrant { token_sha256: sha_hex(t), grant: grant.to_string(), joules_ceiling_micro: joules, funds_ceiling: u64::MAX, currency: String::new() };
let server = relay(|c| {
c.auth.required = true;
c.auth.grants = vec![by_proof(&room_a, &in_a), by_proof(&room_b, &in_b), rule(&[&rb], publisher_token, true, false), rule(&[&rc], room_c_token, true, true)];
c.metering.grants = vec![maps(&in_a, &digest_hex(&in_a), 1_000_000), maps(&in_b, &digest_hex(&in_b), 2_000_000), maps(room_c_token, &of_c, u64::MAX)];
})
.await;
let relay = server.relay().clone();
let tb = Tracks::new(ns(&[&rb, &hex::encode(performer.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
let tc = Tracks::new(ns(&[&rc, &hex::encode(other.verifying_key().to_bytes()), "video"]), "0", 20, 6, 128);
let (p, _ep) = connect(&server, Via::Quic, bearer(publisher_token)).await;
let _pp = publish_tracks(&p, &tb, None).await;
let (q, _eq) = connect(&server, Via::Quic, bearer(room_c_token)).await;
let _pq = publish_tracks(&q, &tc, None).await;
let holds = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: room_c_token.to_vec() }], ..proving(&server, &[(&in_a, &viewer), (&in_b, &viewer)]) };
let (v, _) = connect(&server, Via::WebSocket, holds).await;
let sb = subscribe_both(&v, &tb, Parameters::new()).await;
let sc = subscribe_both(&v, &tc, Parameters::new()).await;
publish_group(&p, &performer, &tb, 0, &payload("b", 0, 300)).await;
publish_group(&q, &other, &tc, 0, &payload("c", 0, 300)).await;
let got = collect_concurrent(&v, STEP, Duration::from_millis(300), |g| groups_of(g, sb.media_alias) == [0] && groups_of(g, sc.media_alias) == [0]).await;
assert_eq!((groups_of(&got, sb.media_alias), groups_of(&got, sc.media_alias)), (vec![0], vec![0]));
let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
assert!(until(STEP, || forwarded(&relay) == 4).await, "both rooms' objects and receipts counted out");
let grants_of = |rec: &HopRecord| rec.ledger.iter().map(|e| hex::encode(e.grant)).collect::<Vec<_>>();
let holds_to_its_room = |what: &str, b: &HopRecord, c: &HopRecord| {
let names = grants_of(b);
assert!(names.contains(&digest_hex(&in_b)), "{what}: room B's grant: {names:?}");
assert!(!names.contains(&digest_hex(&in_a)) && !names.contains(&of_c), "{what}: another room's grant in room B's ledger: {names:?}");
let e = b.ledger.iter().find(|e| hex::encode(e.grant) == digest_hex(&in_b)).expect("the viewer's grant");
assert_eq!((e.joules_ceiling, e.sessions), (2_000_000, 1), "{what}: its ceiling");
assert!(e.bytes_out > 0 && e.subscriber_us.iter().sum::<u64>() > 0, "{what}: its bytes and its time count against it");
assert_eq!(grants_of(c), vec![of_c.clone()], "{what}: the token's grant, in the room no record names");
assert_eq!(c.ledger[0].sessions, 2, "{what}: its publisher and the viewer");
};
// Over a window of the sessions, still open, and again over the sessions ended.
let w = relay.flush_and_seal(Duration::from_millis(200)).await;
let (b, c) = (relay.window_record(room_b, t0, w.index).unwrap(), relay.window_record(room_c, t0, w.index).unwrap());
holds_to_its_room("a window", &b, &c);
conformance_reads(&b.to_bytes(), &b.ledger_bytes(), &room_b, 1);
conformance_reads(&c.to_bytes(), &c.ledger_bytes(), &room_c, 1);
assert!(relay.window_record(room_a, t0, w.index).unwrap().ledger.is_empty(), "nothing of room A was held");
for x in [&p, &q, &v] {
x.close(0, "done");
}
ended(&relay, 3).await;
holds_to_its_room("the sessions ended", &relay.hop_record(room_b, t0), &relay.hop_record(room_c, t0));
let s = relay.summaries().into_iter().map(|s| s.summary).find(|s| s.tally.room_grants.len() == 2).expect("the viewer's summary names each grant");
assert_eq!(s.tally.room_grants[&hex::encode(room_a)].grant, digest_hex(&in_a));
assert_eq!(s.tally.room_grants[&hex::encode(room_b)].grant, digest_hex(&in_b));
assert_eq!(s.tally.room_grant.map(|g| g.grant), Some(of_c.clone()), "the grant of no room");
server.shutdown().await;
}
// ── A revocation stays ──────────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_revocation_outlives_a_reload_a_set_and_a_restart() {
let _turn = GRANTS_FILE.lock().await;
timeout(Duration::from_secs(180), stays_run()).await.expect("the test finishes in time");
}
async fn stays_run() {
let authority = SigningKey::from_bytes(&[0x64; 32]);
let ak = hex::encode(authority.verifying_key().to_bytes());
let room = [0x7c; 32];
let (k_gone, k_stands) = (SigningKey::from_bytes(&[0x76; 32]), SigningKey::from_bytes(&[0x77; 32]));
let gone = room_grant(&authority, room, &k_gone, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
let stands = room_grant(&authority, room, &k_stands, 0, false, [0; 16], u64::MAX, u64::MAX, &[0; 3]);
let sock = socket_path("stays");
let dir = PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join(format!("stays-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let (grants, kept) = (dir.join("grants.toml"), dir.join("revoked.toml"));
std::fs::write(&grants, grants_file(&[])).unwrap();
let config = |c: &mut RelayConfig| {
c.auth.required = true;
c.auth.authorities = vec![Authority { key: ak.clone(), rooms: vec![] }];
c.control.socket = Some(sock.clone());
c.control.grants_path = Some(grants.clone());
c.control.revocations_path = Some(kept.clone());
};
// The revoked grant closed at its SETUP, and the other standing.
let holds = |server: joule_moq_relay::RelayServer, when: &'static str| {
let (gone, stands, k_gone, k_stands) = (gone.clone(), stands.clone(), k_gone.clone(), k_stands.clone());
async move {
let relay = server.relay().clone();
let target = format!("//{}{}", server.websocket_addr.unwrap(), relay.config().path);
let with = |g: &Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g.clone(), k.clone())], target: target.clone(), ..ClientOptions::default() };
let (c, _) = connect(&server, Via::WebSocket, with(&gone, &k_gone)).await;
let why = closed_with(&c, SessionError::UNAUTHORIZED, when).await;
assert!(why.contains("grant_revoked"), "{when}: {why}");
let (c, _) = connect(&server, Via::WebSocket, with(&stands, &k_stands)).await;
assert!(until(STEP, || relay.stats().sessions_open == 1).await, "{when}: the other grant stands");
c.close(0, "done");
assert!(until(STEP, || relay.stats().sessions_open == 0).await, "{when}");
server
}
};
let server = relay(config).await;
let x = ask(&sock, json!({"op": "revoke", "authority": ak, "revoked": [digest_hex(&gone)]})).await;
assert_eq!(x["ok"], true, "{x}");
let server = holds(server, "revoked").await;
// A reload of a file that does not name it, and a set that does not repeat it.
let x = ask(&sock, json!({"op": "reload"})).await;
assert_eq!(x["ok"], true, "{x}");
assert_eq!(server.relay().auth_config().revocations.len(), 1, "the list is in force after a reload");
let server = holds(server, "after a reload").await;
let x = ask(&sock, json!({"op": "set", "auth": {"required": true, "grants": [], "authorities": [{"key": ak}]}, "metered": []})).await;
assert_eq!(x["ok"], true, "{x}");
let server = holds(server, "after a set that does not repeat it").await;
// A restart reads the file it kept it in.
server.shutdown().await;
let server = relay(config).await;
let server = holds(server, "after a restart").await;
server.shutdown().await;
// A file that does not read stops the relay: it never starts without what it was told to keep.
std::fs::write(&kept, "[[revocations]]\nauthority = ").unwrap();
let mut c = RelayConfig::loopback();
config(&mut c);
assert!(RelayServer::start(c).await.is_err(), "a kept list that does not read");
std::fs::remove_dir_all(&dir).unwrap();
}
// ── A flush inside a group ──────────────────────────────────────────────────
/// A Python 3 that runs the conformance verifier, if this host has one. Under CI, and wherever
/// `JOULE_MOQ_RELAY_VERIFY_REQUIRE` is set, one is needed, and a skip is never a pass.
fn python() -> Option<&'static str> {
let found = ["python3", "python3.14", "python3.13", "python3.12", "python3.11", "python3.10", "python3.9", "python3.8"]
.into_iter()
.find(|p| std::process::Command::new(p).args(["-c", "import sys; sys.exit(0 if sys.version_info >= (3, 8) else 1)"]).output().is_ok_and(|o| o.status.success()));
if found.is_none() {
for required in ["JOULE_MOQ_RELAY_VERIFY_REQUIRE", "CI"] {
assert!(std::env::var_os(required).is_none(), "{required} is set, and this host has no Python 3.8+ for live_metering_verify.py");
}
}
found
}
/// Each reading its own files: the tests of this file run at once.
static READINGS: AtomicU64 = AtomicU64::new(0);
/// The record and its ledger read as this crate reads them, with §3.6's rules, and as
/// `live_metering_verify.py` reads them: its layout, class, rules, room and signature, the
/// ledger's entries and root against the record's, and the size bound of its version with
/// `media` media tracks.
fn conformance_reads(record: &[u8], ledger: &[u8], room: &[u8; 32], media: u64) {
HopRecord::decode(record, ledger).expect("the record keeps live-metering §3.6");
let Some(py) = python() else {
eprintln!("SKIPPED: no Python 3.8+ here to run live_metering_verify.py");
return;
};
let dir = PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join("live_control");
std::fs::create_dir_all(&dir).unwrap();
let n = READINGS.fetch_add(1, Ordering::Relaxed);
let (rp, lp) = (dir.join(format!("window-{}-{n}.wmh", std::process::id())), dir.join(format!("window-{}-{n}.wml", std::process::id())));
std::fs::write(&rp, record).unwrap();
std::fs::write(&lp, ledger).unwrap();
let verifiers = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/live-conformance").canonicalize().expect("wai/live-conformance");
let script = r#"
import sys
sys.path.insert(0, sys.argv[1])
import live_metering_verify as V
rec, led, room, m = open(sys.argv[2], "rb").read(), open(sys.argv[3], "rb").read(), bytes.fromhex(sys.argv[4]), int(sys.argv[5])
if len(rec) > V.bound_of(rec, m):
sys.exit("record_too_large")
try:
x = V.read_hop(rec, room)
entries = V.read_ledger(led)
except V.Refuse as e:
sys.exit("refused: %s" % e)
if len(entries) != x["grants"] or V.ledger_root(entries) != x["ledger"]:
sys.exit("ledger_mismatch")
print("read: version %d, %d entries, %d grants, span %d..%d" % (x["version"], len(x["tracks"]), x["grants"], x["start"], x["end"]))
"#;
let out = std::process::Command::new(py).arg("-c").arg(script).arg(&verifiers).arg(&rp).arg(&lp).arg(hex::encode(room)).arg(media.to_string()).output().expect("python runs");
let said = format!("{}{}", String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr));
let _ = (std::fs::remove_file(&rp), std::fs::remove_file(&lp));
assert!(out.status.success(), "live_metering_verify.py refused the record: {said}");
eprintln!("live_metering_verify.py: {}", said.trim());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_flush_inside_a_group_cuts_at_its_boundary() {
timeout(Duration::from_secs(180), flush_run()).await.expect("the test finishes in time");
}
/// Subgroup `sub` of group `g`, objects `first..first + n`, each `len` bytes.
async fn send(c: &Client, alias: u64, g: u64, sub: u64, first: u64, n: u64, last: bool) {
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Explicit(sub), publisher_priority: Some(128), properties: false, end_of_group: last, first_object: true };
let objects: Vec<SubgroupObject> = (first..first + n).map(|o| SubgroupObject::normal(o, payload("v", g * 100 + o, 500))).collect();
c.send_subgroup(h, &objects).await.expect("a subgroup");
}
/// The objects of `alias` read, counted.
fn objects_of(got: &[Got], alias: u64) -> usize {
got.iter().filter(|x| x.header.track_alias == alias).map(|x| x.objects.len()).sum()
}
async fn flush_run() {
let t0 = now_us() - 1_000_000;
let room = [0x3c; 32];
let r = hex::encode(room);
let p = hex::encode([0x4d; 32]);
let sock = socket_path("flush");
let server = relay(|c| {
c.control.socket = Some(sock.clone());
c.control.flush_grace_ms = 200;
})
.await;
let relay = server.relay().clone();
let track_ns = ns(&[&r, &p, "video"]);
let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
let (subscriber, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let (ok, _stx, _srx) = subscriber.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
let alias = ok.track_alias;
// Group 0's first subgroup, before the cut.
send(&publisher, 10, 0, 0, 0, 3, false).await;
let got = collect_concurrent(&subscriber, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 3).await;
assert_eq!(objects_of(&got, alias), 3);
// The cut, inside group 0. Its seal waits for what follows to be read and counted, however
// long a loaded host takes, as a grace would on a host that keeps up.
assert_eq!(relay.flush().0, 0);
// Group 0's second subgroup arrives after the cut; group 1 begins after it.
send(&publisher, 10, 0, 1, 3, 3, true).await;
send(&publisher, 10, 1, 0, 0, 3, true).await;
let got = collect_concurrent(&subscriber, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 6).await;
assert_eq!(objects_of(&got, alias), 6, "group 0's second subgroup and group 1 delivered");
let forwarded = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
assert!(until(STEP, || forwarded(&relay) == 9).await, "the nine objects counted out");
assert_eq!(relay.seal(0), vec![0]);
// Window 0 holds group 0 whole, in and out, with both sessions open.
let w0 = relay.window(0).expect("window 0 is sealed");
let in_of = |w: &joule_moq_relay::relay::Window| w.parts.iter().flat_map(|s| s.tally.publisher.tracks.values()).map(|t| t.objects_received).sum::<u64>();
let out_of = |w: &joule_moq_relay::relay::Window| w.parts.iter().flat_map(|s| s.tally.subscriber.tracks.values()).map(|t| t.objects_forwarded).sum::<u64>();
assert_eq!((in_of(&w0), out_of(&w0)), (6, 6), "group 0 counted whole in the window it began in");
assert_eq!(relay.stats().sessions_open, 2, "nobody left");
// The record over window 0, signed on the control socket, its sessions still open.
let a = ask(&sock, json!({"op": "record", "room": r, "t0_unix_us": t0, "window": 0})).await;
assert_eq!(a["ok"], true, "{a}");
let (bytes, ledger) = (hex::decode(a["record"].as_str().unwrap()).unwrap(), hex::decode(a["ledger"].as_str().unwrap()).unwrap());
// Read first as WAI's second implementation reads a record, then as this crate does.
conformance_reads(&bytes, &ledger, &room, 1);
let rec = HopRecord::decode(&bytes, &ledger).expect("the record reads");
rec.verify().expect("signed by the relay");
assert_eq!(rec.hop, relay.relay_id());
assert_eq!((rec.start, rec.end), (w0.from_unix_us - t0, w0.to_unix_us - t0), "the window's span on the room clock");
let (inn, out): (Vec<&joule_moq_relay::metering::Entry>, Vec<&joule_moq_relay::metering::Entry>) = rec.entries.iter().partition(|e| e.dir == 1);
assert_eq!((inn.len(), out.len(), inn[0].objects, out[0].objects, inn[0].kind), (1, 1, 6, 6, 1));
assert_eq!(rec.sessions, 1, "the subscriber, open");
assert!(rec.subscriber_us[0] > 0 && rec.subscriber_us[1..].iter().all(|&x| x == 0));
assert_eq!(rec.ledger.len(), 1);
assert_eq!((rec.ledger[0].sessions, rec.ledger[0].subscriber_us), (2, rec.subscriber_us));
// A window not sealed is refused.
let a = ask(&sock, json!({"op": "record", "room": r, "t0_unix_us": t0, "window": 7})).await;
assert_eq!(a["ok"], false, "{a}");
// Both sessions end; window 1 is closed and sealed; each session's windows sum to it.
publisher.close(0, "done");
subscriber.close(0, "done");
assert!(until(STEP, || relay.summaries().len() == 2).await);
// The surface's flush closes window 1, after its grace, its sessions ended.
let flushed = ask(&sock, json!({"op": "flush"})).await;
assert_eq!((flushed["ok"].as_bool(), flushed["window"].as_u64(), flushed["sessions"].as_u64()), (Some(true), Some(1), Some(2)), "{flushed}");
let w1 = relay.window(1).expect("window 1");
assert_eq!((in_of(&w1), out_of(&w1)), (3, 3), "group 1 in the window it began in");
for whole in relay.summaries().into_iter().map(|s| s.summary) {
let parts: Vec<&SessionSummary> = w0.parts.iter().chain(w1.parts.iter()).filter(|p| p.session == whole.session).collect();
assert_eq!(parts.len(), 2, "session {}: one part a window", whole.session);
let t = &whole.tally;
let sum = |f: &dyn Fn(&SessionSummary) -> u64| parts.iter().map(|p| f(p)).sum::<u64>();
assert_eq!(sum(&|p| p.tally.subscriber.objects_forwarded), t.subscriber.objects_forwarded);
assert_eq!(sum(&|p| p.tally.subscriber.object_bytes_forwarded), t.subscriber.object_bytes_forwarded);
assert_eq!(sum(&|p| p.tally.subscriber.wire_bytes_sent), t.subscriber.wire_bytes_sent);
assert_eq!(sum(&|p| p.tally.subscriber.undelivered_at_close), t.subscriber.undelivered_at_close);
assert_eq!(sum(&|p| p.tally.publisher.objects_received), t.publisher.objects_received);
assert_eq!(sum(&|p| p.tally.publisher.object_bytes_received), t.publisher.object_bytes_received);
for (n, x) in &t.subscriber.tracks {
let of = |f: &dyn Fn(&joule_moq_relay::summary::TrackTally) -> u64| parts.iter().filter_map(|p| p.tally.subscriber.tracks.get(n)).map(f).sum::<u64>();
assert_eq!((of(&|y| y.objects_forwarded), of(&|y| y.object_bytes_forwarded), of(&|y| y.dropped_delivery_timeout), of(&|y| y.undelivered_at_close)), (x.objects_forwarded, x.object_bytes_forwarded, x.dropped_delivery_timeout, x.undelivered_at_close), "{n}");
}
for (n, x) in &t.publisher.tracks {
let of = |f: &dyn Fn(&joule_moq_relay::summary::PublisherTrackTally) -> u64| parts.iter().filter_map(|p| p.tally.publisher.tracks.get(n)).map(f).sum::<u64>();
assert_eq!((of(&|y| y.objects_received), of(&|y| y.object_bytes_received)), (x.objects_received, x.object_bytes_received), "{n}");
}
let us = |s: &[joule_moq_relay::summary::LayerSpan]| s.iter().map(|x| x.to_unix_us - x.from_unix_us).sum::<u64>();
assert_eq!(sum(&|p| us(&p.tally.subscriber.layer_spans)), us(&t.subscriber.layer_spans), "time on the room's tracks");
}
let total = |f: &dyn Fn(&HopRecord) -> u64| f(&relay.window_record(room, t0, 0).unwrap()) + f(&relay.window_record(room, t0, 1).unwrap());
let whole = relay.hop_record(room, t0);
assert_eq!(total(&|x| x.subscriber_us.iter().sum::<u64>()), whole.subscriber_us.iter().sum::<u64>(), "the windows' records sum to the whole's");
assert_eq!(total(&|x| x.entries.iter().map(|e| e.objects).sum::<u64>()), whole.entries.iter().map(|e| e.objects).sum::<u64>());
assert!(relay.window_record(room, t0, 0).unwrap().end <= relay.window_record(room, t0, 1).unwrap().start, "the windows' spans do not overlap");
server.shutdown().await;
}
// ── A first flush ───────────────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_holds_no_second_copy_of_its_counts_until_its_first_flush() {
timeout(Duration::from_secs(120), first_flush_run()).await.expect("the test finishes in time");
}
async fn first_flush_run() {
let t0 = now_us() - 1_000_000;
let room = [0x3e; 32];
let server = relay(|_| {}).await;
let relay = server.relay().clone();
let track_ns = ns(&[&hex::encode(room), &hex::encode([0x4f; 32]), "video"]);
let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
let (left, _el) = connect(&server, Via::Quic, ClientOptions::default()).await;
let (ok_left, _ltx, _lrx) = left.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
let (stays, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
let (ok_stays, _stx, _srx) = stays.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
send(&publisher, 10, 0, 0, 0, 3, true).await;
for (c, alias) in [(&left, ok_left.track_alias), (&stays, ok_stays.track_alias)] {
let got = collect_concurrent(c, STEP, Duration::from_millis(200), |g| objects_of(g, alias) == 3).await;
assert_eq!(objects_of(&got, alias), 3);
}
let counted_out = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.subscriber.objects_forwarded).sum::<u64>();
assert!(until(STEP, || counted_out(&relay) == 6).await, "both subscribers' objects counted out");
left.close(0, "done");
assert!(until(STEP, || relay.summaries().len() == 1).await);
// Never flushed: every session's counts are its own whole, held once.
assert_eq!(relay.ended_parts(), 0, "an ended session's counts are its summary");
assert!(relay.open_tallies().iter().all(|(_, t)| t.parts.by_window.is_empty()), "an open session's are its tally");
assert!(relay.window(0).is_none(), "nothing seals by itself");
// The first flush: window 0 holds what the relay counted before it, ended and open.
let w0 = relay.flush_and_seal(Duration::from_millis(100)).await;
assert_eq!((w0.index, w0.parts.len()), (0, 3), "the session ended and the two open");
let ended = relay.summaries().remove(0).summary;
let part = w0.parts.iter().find(|p| p.session == ended.session).expect("the ended session's part");
assert_eq!(part.tally, ended.tally, "its whole, the only window it was in");
assert_eq!(part.close_code, None, "a part states no close");
let count = |w: &joule_moq_relay::relay::Window| (w.parts.iter().map(|p| p.tally.publisher.objects_received).sum::<u64>(), w.parts.iter().map(|p| p.tally.subscriber.objects_forwarded).sum::<u64>());
assert_eq!(count(&w0), (3, 6), "the publisher's three objects, sent to two subscribers");
let rec = relay.window_record(room, t0, 0).expect("the record over window 0");
conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
assert_eq!(rec.sessions, 2, "the subscriber ended and the one open");
assert_eq!(relay.ended_parts(), 0, "sealed");
// From then on the relay counts by window.
send(&publisher, 10, 1, 0, 0, 3, true).await;
let got = collect_concurrent(&stays, STEP, Duration::from_millis(200), |g| objects_of(g, ok_stays.track_alias) == 3).await;
assert_eq!(objects_of(&got, ok_stays.track_alias), 3);
assert!(until(STEP, || counted_out(&relay) == 6).await, "the three of the group after, on the three before");
assert!(relay.open_tallies().iter().any(|(_, t)| !t.parts.by_window.is_empty()), "kept by window now");
publisher.close(0, "done");
stays.close(0, "done");
assert!(until(STEP, || relay.summaries().len() == 3).await);
assert_eq!(relay.ended_parts(), 2, "the two sessions that ended since, held for the window's seal");
let w1 = relay.flush_and_seal(Duration::from_millis(100)).await;
assert_eq!(count(&w1), (3, 3));
for whole in relay.summaries().into_iter().map(|s| s.summary) {
let parts: Vec<&SessionSummary> = w0.parts.iter().chain(w1.parts.iter()).filter(|p| p.session == whole.session).collect();
let sum = |f: &dyn Fn(&SessionSummary) -> u64| parts.iter().map(|p| f(p)).sum::<u64>();
assert_eq!(sum(&|p| p.tally.subscriber.objects_forwarded), whole.tally.subscriber.objects_forwarded, "session {}", whole.session);
assert_eq!(sum(&|p| p.tally.publisher.objects_received), whole.tally.publisher.objects_received, "session {}", whole.session);
let us = |s: &[joule_moq_relay::summary::LayerSpan]| s.iter().map(|x| x.to_unix_us - x.from_unix_us).sum::<u64>();
assert_eq!(sum(&|p| us(&p.tally.subscriber.layer_spans)), us(&whole.tally.subscriber.layer_spans), "session {}: time on the room's tracks", whole.session);
}
server.shutdown().await;
}
// ── An interval ─────────────────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_relay_given_an_interval_seals_with_nobody_asking() {
timeout(Duration::from_secs(120), interval_run()).await.expect("the test finishes in time");
}
async fn interval_run() {
// A ceiling is held at a seal, so a relay told to hold one needs an interval to seal by.
let mut c = RelayConfig::loopback();
c.control.hold_joule_ceilings = true;
let e = RelayServer::start(c).await.err().expect("a relay that would hold no ceiling started");
assert!(e.contains("flush_interval_ms"), "{e}");
let room = [0x3f; 32];
let server = relay(|c| {
c.control.flush_interval_ms = Some(100);
c.control.flush_grace_ms = 50;
})
.await;
let relay = server.relay().clone();
let track_ns = ns(&[&hex::encode(room), &hex::encode([0x50; 32]), "video"]);
let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
send(&publisher, 10, 0, 0, 0, 3, true).await;
let sealed = |relay: &joule_moq_relay::Relay| (0..256).filter_map(|w| relay.window(w)).flat_map(|w| w.parts).map(|p| p.tally.publisher.objects_received).sum::<u64>();
assert!(until(STEP, || sealed(&relay) == 3).await, "the objects stand in a sealed window, and no one flushed");
assert_eq!(relay.stats().sessions_open, 1, "nobody left");
publisher.close(0, "done");
server.shutdown().await;
}
// ── A window with a DVR store ───────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_windows_store_figures_stay_within_its_in_entries() {
timeout(Duration::from_secs(180), dvr_window_run()).await.expect("the test finishes in time");
}
async fn dvr_window_run() {
let t0 = now_us() - 1_000_000;
let room = [0x3d; 32];
let r = hex::encode(room);
// In memory where the host has it (`store_dir`): `written` waits for the store, not a disk.
let dir = store_dir("dvr-window");
let store = dir.clone();
let server = relay(move |c| c.dvr = Some(DvrConfig { path: store, window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 600, max_bytes: 1 << 30 }], room_keys: vec![], takedown_keys: vec![] })).await;
let relay = server.relay().clone();
let track_ns = ns(&[&r, &hex::encode([0x4e; 32]), "video"]);
let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
let _held = publisher.publish(track_ns.clone(), name("0"), 10, Parameters::new(), track_props(128, None)).await.expect("PUBLISH");
let written = |n: u64| {
let relay = relay.clone();
async move { assert!(until(STEP, || relay.dvr().unwrap().figures(room, 0, now_us()).objects_stored == n).await, "{n} objects written") }
};
let came_in = |relay: &joule_moq_relay::Relay| relay.open_tallies().iter().map(|(_, t)| t.publisher.objects_received).sum::<u64>();
// The figures of the record over window `w`, read as the verifier reads it: objects in, and
// objects stored.
let read = |w: u64| {
let rec = relay.window_record(room, t0, w).expect("the window's record");
conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
let inn: u64 = rec.entries.iter().filter(|e| e.dir == 1).map(|e| e.objects).sum();
let d = rec.dvr.expect("version 2");
assert_eq!(d.bytes_stored, d.objects_stored * 500, "window {w}: a payload's bytes");
(inn, d.objects_stored)
};
// A group cut by the flush, its rest stored after the cut and before the seal: the window
// that counted it, in which it began, states it whole.
send(&publisher, 10, 0, 0, 0, 3, false).await;
written(3).await;
relay.flush();
send(&publisher, 10, 0, 1, 3, 3, true).await;
written(6).await;
assert_eq!(relay.seal(0), vec![0]);
assert_eq!(read(0), (6, 6), "the group whole in the window it began in");
// Nothing more: the next window states nothing, and owes nothing.
relay.flush();
assert_eq!(relay.seal(1), vec![1]);
assert_eq!(read(1), (0, 0));
// A group sealed before it is whole, its later subgroup first: nothing of it is stored yet,
// and the window that counted three of its objects states none.
send(&publisher, 10, 1, 1, 3, 3, false).await;
assert!(until(STEP, || came_in(&relay) == 9).await);
relay.flush();
assert_eq!(relay.seal(2), vec![2]);
assert_eq!(read(2), (3, 0), "nothing written yet");
// Its rest arrives, and the store writes all six: the window that counted three states three.
send(&publisher, 10, 1, 0, 0, 3, true).await;
written(12).await;
relay.flush();
assert_eq!(relay.seal(3), vec![3]);
assert_eq!(read(3), (3, 3), "at most what its in entries count");
// A window with no objects in states none, whatever the store has yet to state.
relay.flush();
assert_eq!(relay.seal(4), vec![4]);
assert_eq!(read(4), (0, 0));
// The next with objects in goes on stating it, as far as its own in entries go.
send(&publisher, 10, 2, 0, 0, 3, true).await;
written(15).await;
relay.flush();
assert_eq!(relay.seal(5), vec![5]);
assert_eq!(read(5), (3, 3));
// A window whose own group is not whole yet states what earlier windows could not.
send(&publisher, 10, 3, 1, 3, 3, false).await;
assert!(until(STEP, || came_in(&relay) == 18).await);
relay.flush();
assert_eq!(relay.seal(6), vec![6]);
assert_eq!(read(6), (3, 3), "what the store wrote and no window could state");
// A record over the sessions that ended, its publisher still open: no in entries, so none of
// the store's writes in its span are among them (§3.9).
let (viewer, _ev) = connect(&server, Via::Quic, ClientOptions::default()).await;
let (ok, _vtx, _vrx) = viewer.subscribe(track_ns.clone(), name("0"), Parameters::new()).await.expect("SUBSCRIBE");
send(&publisher, 10, 4, 0, 0, 3, true).await;
written(18).await;
let got = collect_concurrent(&viewer, STEP, Duration::from_millis(200), |g| objects_of(g, ok.track_alias) == 3).await;
assert_eq!(objects_of(&got, ok.track_alias), 3);
viewer.close(0, "done");
assert!(until(STEP, || relay.summaries().len() == 1).await);
let rec = relay.hop_record(room, t0);
conformance_reads(&rec.to_bytes(), &rec.ledger_bytes(), &room, 1);
assert_eq!(rec.dvr.expect("version 2").objects_stored, 0, "none of its in entries count them");
publisher.close(0, "done");
server.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
// ── SIGHUP ──────────────────────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sighup_reads_the_grants_file_again() {
let _turn = GRANTS_FILE.lock().await;
timeout(Duration::from_secs(180), sighup_run()).await.expect("the test finishes in time");
}
fn grants_file(rules: &[(&[u8], bool, bool)]) -> String {
rules.iter().map(|(t, p, s)| format!("[[grants]]\nprefix = [\"live\"]\ntoken_sha256 = \"{}\"\npublish = {p}\nsubscribe = {s}\n\n", sha_hex(t))).collect()
}
async fn sighup_run() {
let (z, x, y) = (b"publisher-token".as_slice(), b"first-viewer-token".as_slice(), b"second-viewer-token".as_slice());
let file = std::env::temp_dir().join(format!("wai-relay-grants-{}.toml", std::process::id()));
std::fs::write(&file, grants_file(&[(z, true, false), (x, false, true)])).unwrap();
let server = relay(|c| {
c.auth.required = true;
c.control.grants_path = Some(file.clone());
})
.await;
let relay = server.relay().clone();
let t = Tracks::new(ns(&["live", "s"]), "cam", 10, 4, 128);
let key = SigningKey::from_bytes(&[0xc3; 32]);
let (p, _ep) = connect(&server, Via::Quic, bearer(z)).await;
let _pp = publish_tracks(&p, &t, None).await;
let (vx, _ex) = connect(&server, Via::Quic, bearer(x)).await;
let _sx = subscribe_both(&vx, &t, Parameters::new()).await;
let (vy, _ey) = connect(&server, Via::Quic, bearer(y)).await;
assert!(vy.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.is_err(), "not in the file yet");
std::fs::write(&file, grants_file(&[(z, true, false), (y, false, true)])).unwrap();
// SAFETY: kill with this process's own id and a signal the relay handles.
assert_eq!(unsafe { libc::kill(libc::getpid(), libc::SIGHUP) }, 0);
assert!(until(STEP, || relay.auth_config().grants.iter().any(|g| g.token_sha256 == sha_hex(y))).await, "the file read again");
let why = closed_with(&vx, SessionError::UNAUTHORIZED, "the viewer whose token the file dropped").await;
assert!(why.starts_with("grant_revoked"), "{why}");
let sy = subscribe_both(&vy, &t, Parameters::new()).await;
publish_group(&p, &key, &t, 0, &payload("s", 0, 300)).await;
let got = collect_concurrent(&vy, STEP, Duration::from_millis(300), |g| groups_of(g, sy.media_alias) == [0]).await;
assert_eq!(groups_of(&got, sy.media_alias), vec![0], "the token the file added is served, by the publisher it kept");
let (vx2, _ex2) = connect(&server, Via::Quic, bearer(x)).await;
assert!(vx2.subscribe(t.ns.clone(), t.media.clone(), Parameters::new()).await.is_err(), "the dropped token refused");
// A file that does not read changes nothing.
std::fs::write(&file, "[[grants]]\nprefix = 7\n").unwrap();
assert!(relay.reload_grants().is_err());
assert!(relay.auth_config().grants.iter().any(|g| g.token_sha256 == sha_hex(y)));
for c in [&p, &vy, &vx2] {
c.close(0, "done");
}
let _ = std::fs::remove_file(&file);
server.shutdown().await;
}
Reference in a new issue View git blame Copy permalink