283 lines
14 KiB
Rust
283 lines
14 KiB
Rust
//! The floor of a WAI room (WAI co-performance §6), through a relay on loopback that holds
|
|
//! it: the room's clock publishes a turn schedule that refuses on its events track, and two
|
|
//! performers each publish an audio track that one subscriber follows. The object of the
|
|
//! performer whose turn it is reaches the subscriber, and the other's does not; once the first
|
|
//! hands its turn to the second on its own events track, the second's reaches the subscriber
|
|
//! and the first's does not. Each refusal is counted.
|
|
//!
|
|
//! And the room's audience inputs (§7.3), through a relay that honours a room authority's
|
|
//! grants: the clock opens a vote on its events track; a member whose grant carries the
|
|
//! audience-input power sends one vote that a subscriber of its events track receives, and a
|
|
//! second, past its power, that the relay withholds; a viewer whose grant carries no such power
|
|
//! has its vote withheld; each refusal is counted in its session's summary by reason.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::{Signer, SigningKey};
|
|
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::Parameters;
|
|
use joule_moq::TrackNamespace;
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{Authority, FloorConfig, Grant};
|
|
use joule_moq_relay::floor::Place;
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
use tokio::time::timeout;
|
|
|
|
fn now_us() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
|
|
}
|
|
|
|
fn hexkey(k: &SigningKey) -> String {
|
|
hex::encode(k.verifying_key().to_bytes())
|
|
}
|
|
|
|
/// A `meta` record of live-events §3.1, without an anchor, signed by `key`.
|
|
fn meta(key: &SigningKey, room: [u8; 32], seq: u64, t: u64, payload: &[u8]) -> Vec<u8> {
|
|
let mut b = b"WEV1\x01".to_vec();
|
|
b.extend_from_slice(&room);
|
|
b.extend_from_slice(&key.verifying_key().to_bytes());
|
|
b.extend_from_slice(&seq.to_le_bytes());
|
|
b.extend_from_slice(&[0; 32]);
|
|
b.extend_from_slice(&t.to_le_bytes());
|
|
b.push(0);
|
|
b.push(0x01);
|
|
b.extend_from_slice(&(payload.len() as u16).to_le_bytes());
|
|
b.extend_from_slice(payload);
|
|
let mut m = b"wai:live-event\x01".to_vec();
|
|
m.extend_from_slice(&b);
|
|
b.extend_from_slice(&key.sign(&m).to_bytes());
|
|
b
|
|
}
|
|
|
|
fn msf(record: &[u8]) -> Vec<u8> {
|
|
use base64::Engine;
|
|
format!("[{{\"t\":1,\"data\":{{\"wev1\":\"{}\"}}}}]", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(record)).into_bytes()
|
|
}
|
|
|
|
/// A `WPT1` schedule (co-performance §6.1) under refuse: from `from`, one segment of two
|
|
/// turns of `len` µs, held by `holders` in turn.
|
|
fn schedule(t0: u64, from: u64, len: u32, holders: &[&SigningKey]) -> Vec<u8> {
|
|
let mut b = b"WPT1\x01".to_vec();
|
|
b.extend_from_slice(&t0.to_le_bytes());
|
|
b.extend_from_slice(&from.to_le_bytes());
|
|
b.push(1);
|
|
b.extend_from_slice(&0u32.to_le_bytes());
|
|
b.push(0);
|
|
b.push(1);
|
|
b.extend_from_slice(&from.to_le_bytes());
|
|
b.extend_from_slice(&len.to_le_bytes());
|
|
b.extend_from_slice(&2u16.to_le_bytes());
|
|
b.push(0);
|
|
b.push(holders.len() as u8);
|
|
for h in holders {
|
|
b.extend_from_slice(&h.verifying_key().to_bytes());
|
|
}
|
|
b
|
|
}
|
|
|
|
/// A `WPH1` handover (co-performance §6.3).
|
|
fn handover(at: u64, to: &SigningKey) -> Vec<u8> {
|
|
let mut b = b"WPH1\x01".to_vec();
|
|
b.extend_from_slice(&at.to_le_bytes());
|
|
b.extend_from_slice(&to.verifying_key().to_bytes());
|
|
b
|
|
}
|
|
|
|
/// One object of group `g`, on a stream of its own.
|
|
async fn send(c: &Client, alias: u64, g: u64, payload: &[u8]) {
|
|
let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(h, &[SubgroupObject::normal(0, payload.to_vec())]).await.expect("an object");
|
|
}
|
|
|
|
fn track(room: [u8; 32], who: &SigningKey, field: &str) -> TrackNamespace {
|
|
ns(&[&hex::encode(room), &hexkey(who), field])
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_holds_the_floor() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let room = [0x61; 32];
|
|
let clock = SigningKey::from_bytes(&[0x62; 32]);
|
|
let (a, b) = (SigningKey::from_bytes(&[0x63; 32]), SigningKey::from_bytes(&[0x64; 32]));
|
|
let server = relay(|c| c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)] })).await;
|
|
let relay = server.relay().clone();
|
|
let floor = || relay.floor().expect("the relay holds the floor");
|
|
// The room began ten seconds ago; its clock gives the floor to a, then b, ten minutes each.
|
|
let t0 = now_us() - 10_000_000;
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
let rc = now_us() - t0;
|
|
send(&c, 1, 0, &msf(&meta(&clock, room, 0, rc, &schedule(t0, rc, 600_000_000, &[&a, &b])))).await;
|
|
let place = |who: &SigningKey| floor().place(&track(room, who, "audio"), now_us());
|
|
assert!(until(STEP, || place(&a) == Some(Place::InTurn)).await, "the relay reads the schedule");
|
|
assert_eq!(place(&b), Some(Place::OutOfTurn(1)));
|
|
|
|
// Both performers publish audio; one subscriber follows both.
|
|
let (pa, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pb, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ha = pa.publish(track(room, &a, "audio"), name("0"), 10, Parameters::new(), track_props(0, None)).await.expect("PUBLISH a's audio");
|
|
let _hb = pb.publish(track(room, &b, "audio"), name("0"), 20, Parameters::new(), track_props(0, None)).await.expect("PUBLISH b's audio");
|
|
let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _sa = s.subscribe(track(room, &a, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE a's audio");
|
|
let _sb = s.subscribe(track(room, &b, "audio"), name("0"), Parameters::new()).await.expect("SUBSCRIBE b's audio");
|
|
send(&pa, 10, 1, b"a in its turn").await;
|
|
send(&pb, 20, 1, b"b out of turn").await;
|
|
let got = collect(&s, Duration::from_secs(3), |g| g.len() >= 2).await;
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![b"a in its turn".as_slice()], "only the object in turn is forwarded");
|
|
assert_eq!(floor().refused(), 1);
|
|
|
|
// a hands its turn to b, from now.
|
|
let _ae = pa.publish(track(room, &a, "events"), name("0"), 11, Parameters::new(), track_props(16, None)).await.expect("PUBLISH a's events");
|
|
let at = now_us() - t0;
|
|
send(&pa, 11, 0, &msf(&meta(&a, room, 0, at, &handover(at, &b)))).await;
|
|
assert!(until(STEP, || place(&b) == Some(Place::InTurn)).await, "the relay reads the handover");
|
|
send(&pa, 10, 2, b"a after its handover").await;
|
|
send(&pb, 20, 2, b"b in the turn it was handed").await;
|
|
let got = collect(&s, Duration::from_secs(3), |g| g.len() >= 2).await;
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![b"b in the turn it was handed".as_slice()]);
|
|
assert_eq!(floor().refused(), 2);
|
|
for x in [&c, &pa, &pb, &s] {
|
|
x.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
}
|
|
|
|
/// A grant of `room` for `principal`, as live-rooms §4.1 lays it out, signed by `authority`:
|
|
/// a viewer's (no media, its policy every kind but the lifecycle's at the default bounds),
|
|
/// with `attrs`.
|
|
fn viewer_grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, attrs: &[(&str, &str)]) -> Vec<u8> {
|
|
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
|
|
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(&[1, 0, 1, 0, 0x10, 0]);
|
|
g.extend_from_slice(&[0; 32]);
|
|
g.extend_from_slice(b"WEP1\x01");
|
|
g.extend_from_slice(&[0x0b; 32]);
|
|
g.extend_from_slice(&1024u16.to_le_bytes());
|
|
g.extend_from_slice(&500u16.to_le_bytes());
|
|
g.extend_from_slice(&50u32.to_le_bytes());
|
|
g.extend_from_slice(&1_000_000u64.to_le_bytes());
|
|
g.extend_from_slice(&1u32.to_le_bytes());
|
|
g.extend_from_slice(&1_000_000u64.to_le_bytes());
|
|
g.extend_from_slice(&[0xff; 16]);
|
|
g.extend_from_slice(&[0; 3]);
|
|
g.extend_from_slice(&[0, 0]);
|
|
let mut a = Vec::new();
|
|
for (k, v) in attrs {
|
|
a.push(k.len() as u8);
|
|
a.extend_from_slice(k.as_bytes());
|
|
a.extend_from_slice(&(v.len() as u16).to_le_bytes());
|
|
a.extend_from_slice(v.as_bytes());
|
|
}
|
|
g.extend_from_slice(&(a.len() as u16).to_le_bytes());
|
|
g.extend_from_slice(&a);
|
|
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
|
|
}
|
|
|
|
/// A `meta` record with a `prev`, so a chain of several reads as one.
|
|
fn meta_after(key: &SigningKey, room: [u8; 32], seq: u64, prev: [u8; 32], t: u64, payload: &[u8]) -> Vec<u8> {
|
|
let mut r = meta(key, room, seq, t, payload);
|
|
r[77..109].copy_from_slice(&prev);
|
|
let body = r.len() - 64;
|
|
let mut m = b"wai:live-event\x01".to_vec();
|
|
m.extend_from_slice(&r[..body]);
|
|
let sig = key.sign(&m).to_bytes();
|
|
r[body..].copy_from_slice(&sig);
|
|
r
|
|
}
|
|
|
|
fn sha256(b: &[u8]) -> [u8; 32] {
|
|
ring::digest::digest(&ring::digest::SHA256, b).as_ref().try_into().unwrap()
|
|
}
|
|
|
|
/// A record's digest (live-events §3.2): over all of it but its signature.
|
|
fn digest(record: &[u8]) -> [u8; 32] {
|
|
sha256(&record[..record.len() - 64])
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_relay_refuses_invalid_audience_inputs() {
|
|
timeout(Duration::from_secs(120), inputs()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn inputs() {
|
|
let room = [0x71; 32];
|
|
let authority = SigningKey::from_bytes(&[0x72; 32]);
|
|
let clock = SigningKey::from_bytes(&[0x73; 32]);
|
|
let (member, viewer, watcher) = (SigningKey::from_bytes(&[0x74; 32]), SigningKey::from_bytes(&[0x75; 32]), SigningKey::from_bytes(&[0x76; 32]));
|
|
let token = b"the clock's token".to_vec();
|
|
let token_sha = hex::encode(sha256(&token));
|
|
let server = relay(|c| {
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
|
|
c.auth.grants = vec![Grant { prefix: vec!["*".into(), hexkey(&clock), "events".into()], token_sha256: token_sha.clone(), publish: true, subscribe: true, proof: false, unlimited: true, seek: false }];
|
|
c.floor = Some(FloorConfig { room_keys: vec![hexkey(&clock)] });
|
|
})
|
|
.await;
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let t0 = now_us() - 10_000_000;
|
|
let rc = || now_us() - t0;
|
|
|
|
// The clock opens a vote of three options, closing in a minute.
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions { tokens: vec![joule_moq::draft22::Token::UseValue { token_type: 0, value: token.clone() }], ..ClientOptions::default() }).await;
|
|
let _ce = c.publish(track(room, &clock, "events"), name("0"), 1, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the clock's events");
|
|
let mut w = b"WPW1\x01".to_vec();
|
|
w.extend_from_slice(&[1, 0b100, 3, 1]);
|
|
w.extend_from_slice(&(rc() + 60_000_000).to_le_bytes());
|
|
w.extend_from_slice(&[0; 32]);
|
|
w.extend_from_slice(&0u16.to_le_bytes());
|
|
let wr = meta(&clock, room, 0, rc(), &w);
|
|
let wd = digest(&wr);
|
|
send(&c, 1, 0, &msf(&wr)).await;
|
|
let vote = |o: u8| {
|
|
let mut b = b"WPA1\x01".to_vec();
|
|
b.extend_from_slice(&wd);
|
|
b.extend_from_slice(&[3, o]);
|
|
b
|
|
};
|
|
|
|
// A watcher follows the member's and the viewer's events tracks.
|
|
let (s, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &watcher, &[]), &watcher)).await;
|
|
// The member may send one input to a window; the viewer none.
|
|
let (pm, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &member, &[("audience_input", "1")]), &member)).await;
|
|
let (pv, _) = connect(&server, Via::WebSocket, with(viewer_grant(&authority, room, &viewer, &[]), &viewer)).await;
|
|
let _pme = pm.publish(track(room, &member, "events"), name("0"), 2, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the member's events");
|
|
let _pve = pv.publish(track(room, &viewer, "events"), name("0"), 3, Parameters::new(), track_props(16, None)).await.expect("PUBLISH the viewer's events");
|
|
let _sm = s.subscribe(track(room, &member, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the member's events");
|
|
let _sv = s.subscribe(track(room, &viewer, "events"), name("0"), Parameters::new()).await.expect("SUBSCRIBE the viewer's events");
|
|
tokio::time::sleep(Duration::from_millis(300)).await;
|
|
let first = meta(&member, room, 0, rc(), &vote(1));
|
|
send(&pm, 2, 0, &msf(&first)).await;
|
|
let second = meta_after(&member, room, 1, digest(&first), rc(), &vote(2));
|
|
send(&pm, 2, 1, &msf(&second)).await;
|
|
send(&pv, 3, 0, &msf(&meta(&viewer, room, 0, rc(), &vote(0)))).await;
|
|
let got = collect(&s, Duration::from_secs(3), |g| !g.is_empty() && g.len() >= 3).await;
|
|
let payloads: Vec<&[u8]> = got.iter().flat_map(|g| g.objects.iter().map(|o| o.payload.as_slice())).collect();
|
|
assert_eq!(payloads, vec![msf(&first).as_slice()], "only the member's first vote is forwarded");
|
|
let f = server.relay().floor().unwrap();
|
|
assert_eq!(f.refused_inputs(), 2);
|
|
for x in [&pm, &pv, &s, &c] {
|
|
x.close(0, "done");
|
|
}
|
|
let relay = server.relay().clone();
|
|
assert!(until(STEP, || relay.summaries().len() >= 4).await, "every session's summary");
|
|
let refused: Vec<(String, u64)> = relay.summaries().iter().flat_map(|x| x.summary.tally.publisher.co_refused.clone()).collect();
|
|
let mut refused = refused;
|
|
refused.sort();
|
|
assert_eq!(refused, vec![("input_not_permitted".to_string(), 1), ("input_over_limit".to_string(), 1)]);
|
|
server.shutdown().await;
|
|
}
|