203 lines
11 KiB
Rust
203 lines
11 KiB
Rust
//! A relay that serves a room whose authority issues grants as visitors arrive (WAI
|
|
//! live-rooms §10.2, §10.5): it holds the authority's key, not a rule for each grant.
|
|
//!
|
|
//! - A grant the authority signed, presented in SETUP with its proof, gives the rights
|
|
//! §10.2 maps it onto: a publisher its own media namespaces, a viewer the room to
|
|
//! subscribe to; the same grant with no proof closes the session, and a grant signed by
|
|
//! another key gives nothing.
|
|
//! - The door, with a rule over `["*", "*", "video"]` and `trust_field = 1`, holds each
|
|
//! publisher's video to the key its namespace names: a group signed by another key is
|
|
//! dropped there.
|
|
//! - The relay's limits end a session: at its session limit (`NO_ERROR`), when its grant
|
|
//! expires (`EXPIRED_AUTH_TOKEN`), and past its ingress limit (`INTERNAL_ERROR`).
|
|
//!
|
|
//! Over the WebSocket binding, the transport a deployment behind an HTTP proxy uses.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::{Signer, SigningKey};
|
|
use joule_moq::draft22::{Parameters, RequestErrorCode, SessionError, Token};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{Authority, DoorRule};
|
|
use joule_moq_relay::transport::TransportError;
|
|
use joule_moq_relay::RelayServer;
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
use tokio::time::timeout;
|
|
|
|
fn now_s() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
|
|
}
|
|
|
|
/// A grant laid out as live-rooms §4.1 lays it out: `publish` the media bits, a policy that
|
|
/// permits the chat kind when `events`, signed by `authority`.
|
|
fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, nbf: u64, exp: u64, publish: u8, events: bool) -> Vec<u8> {
|
|
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(&nbf.to_le_bytes());
|
|
g.extend_from_slice(&exp.to_le_bytes());
|
|
g.extend_from_slice(&[0, publish, 1, 0, 0, 0]);
|
|
g.extend_from_slice(&[0; 32]);
|
|
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(&[0xff; 16]);
|
|
g.extend_from_slice(&[0; 3]);
|
|
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 target(server: &RelayServer) -> String {
|
|
format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path)
|
|
}
|
|
|
|
fn with(server: &RelayServer, record: &[u8], key: &SigningKey) -> ClientOptions {
|
|
ClientOptions { grants: vec![(record.to_vec(), key.clone())], target: target(server), ..ClientOptions::default() }
|
|
}
|
|
|
|
async fn closed_with(c: &Client, code: SessionError, within: Duration, what: &str) -> String {
|
|
match timeout(within, 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:?}"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_authoritys_grants_their_proofs_the_door_and_the_limits() {
|
|
timeout(Duration::from_secs(180), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let authority = SigningKey::from_bytes(&[0x61; 32]);
|
|
let stranger = SigningKey::from_bytes(&[0x62; 32]);
|
|
let (alice, bob, viewer) = (SigningKey::from_bytes(&[0x63; 32]), SigningKey::from_bytes(&[0x64; 32]), SigningKey::from_bytes(&[0x65; 32]));
|
|
let room = [0x7e; 32];
|
|
let r = hex::encode(room);
|
|
let server = relay(|c| {
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hex::encode(authority.verifying_key().to_bytes()), rooms: vec![] }];
|
|
c.receipts.door = vec![DoorRule { prefix: vec!["*".into(), "*".into(), "video".into()], trust: vec![], trust_field: Some(1), hold_ms: 2_000 }];
|
|
})
|
|
.await;
|
|
let now = now_s();
|
|
let ga = grant(&authority, room, &alice, now - 60, now + 600, 0b011, true);
|
|
let gb = grant(&authority, room, &bob, now - 60, now + 600, 0b011, true);
|
|
let gv = grant(&authority, room, &viewer, now - 60, now + 600, 0, false);
|
|
let ta = Tracks::new(ns(&[&r, &hex::encode(alice.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
|
|
let tb = Tracks::new(ns(&[&r, &hex::encode(bob.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
|
|
|
|
// Alice publishes her video; the viewer subscribes to it and every object verifies.
|
|
let (a, _) = connect(&server, Via::WebSocket, with(&server, &ga, &alice)).await;
|
|
let _pa = publish_tracks(&a, &ta, None).await;
|
|
let (v, _) = connect(&server, Via::WebSocket, with(&server, &gv, &viewer)).await;
|
|
let sub = subscribe_both(&v, &ta, Parameters::new()).await;
|
|
for g in 0..5 {
|
|
publish_group(&a, &alice, &ta, g, &payload("a", g, 300)).await;
|
|
}
|
|
let got = collect_concurrent(&v, STEP, Duration::from_millis(500), |g| g.iter().filter(|x| x.header.track_alias == sub.media_alias).count() >= 5).await;
|
|
let (objects, verified, _, _) = verify(&got, &sub, &alice);
|
|
assert!(objects >= 5 && verified == objects, "{verified} of {objects} verified");
|
|
|
|
// Alice may not publish under Bob's namespace, the viewer may publish nothing of media,
|
|
// and neither may publish in another room.
|
|
let refused = a.publish(tb.ns.clone(), tb.media.clone(), 30, Parameters::new(), track_props(128, None)).await.expect_err("another principal's namespace");
|
|
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
|
|
let refused = v.publish(ns(&[&r, &hex::encode(viewer.verifying_key().to_bytes()), "video"]), name("0"), 30, Parameters::new(), track_props(128, None)).await.expect_err("a viewer's media");
|
|
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
|
|
|
|
// The door: Bob publishes in his own namespace with Alice's key; his group is dropped.
|
|
let (b, _) = connect(&server, Via::WebSocket, with(&server, &gb, &bob)).await;
|
|
let _pb = publish_tracks(&b, &tb, None).await;
|
|
let sb = subscribe_both(&v, &tb, Parameters::new()).await;
|
|
publish_group(&b, &alice, &tb, 0, &payload("b", 0, 300)).await;
|
|
let got = collect_concurrent(&v, Duration::from_secs(4), Duration::from_millis(500), |_| false).await;
|
|
assert!(got.iter().all(|x| x.header.track_alias != sb.media_alias || x.reset.is_some() || x.objects.is_empty()), "a group signed by another key went past the door");
|
|
assert!(server.relay().stats().refused_at_door >= 1 || server.relay().stats().groups_dropped_at_door >= 1, "the door refused it");
|
|
|
|
// The grant with no proof: the session is refused. A grant another key signed is no
|
|
// grant here: its requests are refused, its session stands.
|
|
let bare = ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: ga.clone() }], ..ClientOptions::default() };
|
|
let (c, _) = connect(&server, Via::WebSocket, bare).await;
|
|
closed_with(&c, SessionError::UNAUTHORIZED, STEP, "a grant with no proof").await;
|
|
let forged = grant(&stranger, room, &alice, now - 60, now + 600, 0b011, true);
|
|
let (c, _) = connect(&server, Via::WebSocket, with(&server, &forged, &alice)).await;
|
|
let refused = c.subscribe(ta.ns.clone(), ta.media.clone(), Parameters::new()).await.expect_err("a grant another key signed");
|
|
assert!(refused.contains(&RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
|
|
for c in [&a, &b, &v, &c] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
|
|
// The limits, each on a relay of its own.
|
|
let limited = |f: fn(&mut joule_moq_relay::RelayConfig)| {
|
|
let key = hex::encode(authority.verifying_key().to_bytes());
|
|
relay(move |c| {
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key, rooms: vec![] }];
|
|
f(c);
|
|
})
|
|
};
|
|
// At its session limit; a session whose token's rule is `unlimited` is not.
|
|
let s = limited(|c| {
|
|
c.limits.max_session_ms = 1_500;
|
|
c.auth.grants = vec![joule_moq_relay::config::Grant { prefix: vec![], token_sha256: hex::encode(joule_moq_relay::auth::sha256(b"service")), publish: false, subscribe: true, proof: false, unlimited: true, seek: false }];
|
|
})
|
|
.await;
|
|
let (c, _) = connect(&s, Via::WebSocket, with(&s, &gv, &viewer)).await;
|
|
let (svc, _) = connect(&s, Via::WebSocket, ClientOptions { tokens: vec![Token::UseValue { token_type: 0, value: b"service".to_vec() }], ..ClientOptions::default() }).await;
|
|
let why = closed_with(&c, SessionError::NO_ERROR, Duration::from_secs(10), "the session limit").await;
|
|
assert!(why.contains("session limit"), "{why}");
|
|
assert!(timeout(Duration::from_secs(2), svc.conn.closed()).await.is_err(), "an unlimited session outlives the limit");
|
|
svc.close(0, "done");
|
|
s.shutdown().await;
|
|
// When its grant expires.
|
|
let s = limited(|_| {}).await;
|
|
let short = grant(&authority, room, &viewer, now - 60, now_s() + 2, 0, false);
|
|
let (c, _) = connect(&s, Via::WebSocket, with(&s, &short, &viewer)).await;
|
|
closed_with(&c, SessionError::EXPIRED_AUTH_TOKEN, Duration::from_secs(10), "the grant's expiry").await;
|
|
s.shutdown().await;
|
|
// Past its ingress limit: 64 kbit/s over a second is 8 kB; ten 4 kB groups pass it.
|
|
let s = limited(|c| {
|
|
c.limits.max_ingress_kbps = 64;
|
|
c.limits.ingress_burst_ms = 1_000;
|
|
})
|
|
.await;
|
|
let (c, _) = connect(&s, Via::WebSocket, with(&s, &ga, &alice)).await;
|
|
let _p = publish_tracks(&c, &ta, None).await;
|
|
for g in 0..10 {
|
|
if c.send_subgroup(
|
|
joule_moq::draft22::data::SubgroupHeader { track_alias: ta.media_alias, group_id: g, subgroup_id: joule_moq::draft22::data::SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true },
|
|
&[joule_moq::draft22::data::SubgroupObject::normal(0, payload("x", g, 4_000))],
|
|
)
|
|
.await
|
|
.is_err()
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
// The relay closes the session while the client may still be writing; then the client
|
|
// sees its pipe break before the close arrives, and the relay's own record says why.
|
|
match timeout(Duration::from_secs(10), c.conn.closed()).await.expect("closed in time") {
|
|
TransportError::Closed { code, reason, .. } => {
|
|
assert_eq!(code, SessionError::INTERNAL_ERROR.0, "{reason}");
|
|
assert!(reason.contains("ingress limit"), "{reason}");
|
|
}
|
|
other => {
|
|
let closed = until(Duration::from_secs(10), || s.relay().summaries().iter().any(|x| x.summary.close_code == Some(SessionError::INTERNAL_ERROR.0))).await;
|
|
assert!(closed, "the relay closed the session past its ingress limit (the client saw {other:?})");
|
|
}
|
|
}
|
|
s.shutdown().await;
|
|
}
|