ffcdf57f9e
Gateway: a challenge store (issued per endpoint or session, consumed, kept for replay detection), joule_moq::grant_proof::check_proof on every request that carries a grant (POST, and PATCH, DELETE, GET and HEAD on session URLs, GET and HEAD on endpoints when a grant is carried), a fresh WAI-Room-Challenge on every response to such a request, the two fields in CORS, admission's single-use check only on a request that opens a session. Receipts tracks renamed *_receipts. Tests: an Auth helper that proves possession; whip_publish, whep_play and codecs moved onto it. auth_and_ice.rs is still the pre-proof version and does not build against the new helpers; room_corpus does not yet run the corpus's proofs. Not built, not tested, not pushed. Co-Authored-By: Claude Opus 5.5 <[email protected]>
74 lines
3.5 KiB
Rust
74 lines
3.5 KiB
Rust
//! A WHIP publisher (str0m as the client) sends synthetic H.264 and Opus into the
|
|
//! gateway. A subscriber at the relay receives every group of both tracks, verifies every
|
|
//! receipt under the gateway's key (and the relay's door verifies them too), and the
|
|
//! frames' bytes equal what was sent. DELETE ends the publish, and the relay ends the
|
|
//! tracks with PUBLISH_DONE.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use joule_moq_ingest::frame::Codec;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_whip_publisher_reaches_a_relay_subscriber_with_every_receipt() {
|
|
timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let w = World::start(|_| {}).await;
|
|
let auth = w.auth(&w.publisher, 0);
|
|
let p = publish(&w, &auth).await;
|
|
let watcher = watch(&w).await;
|
|
|
|
let mut sent = Sent::default();
|
|
send(&p, &mut sent, 3, 30).await;
|
|
assert!(received(&w, &p, sent.video.len() + sent.audio.len()).await, "every frame reached the gateway");
|
|
|
|
// DELETE ends the publish, and the relay sees each track end.
|
|
let r = delete(&w, &p.location, &auth).await;
|
|
assert_eq!(r.status, 200);
|
|
assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
|
|
settle(&watcher, Duration::from_millis(500)).await;
|
|
let r = delete(&w, &p.location, &auth).await;
|
|
assert_eq!(r.status, 404, "the session is gone");
|
|
|
|
let video = watcher.tracks.iter().find(|t| t.kind == 2).unwrap();
|
|
let audio = watcher.tracks.iter().find(|t| t.kind == 1).unwrap();
|
|
let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
|
|
let (af, ag) = verified(&watcher, audio, &w.gateway_key, Codec::Opus);
|
|
assert_eq!(vf.len(), sent.video.len(), "every video frame");
|
|
same(&vf, &sent.video, "video frames byte for byte");
|
|
same(&af, &sent.audio, "audio packets byte for byte");
|
|
assert_eq!(vg, 3, "one video group per keyframe");
|
|
assert!(ag >= 2, "audio groups: {ag}");
|
|
// Each audio group spans less than its 200 ms window on the room clock, and video
|
|
// groups begin at keyframes.
|
|
let mut spans: std::collections::BTreeMap<u64, (u64, u64)> = Default::default();
|
|
for ((alias, g, _), p) in watcher.objects.lock().unwrap().iter() {
|
|
if *alias == audio.media_alias {
|
|
let t = joule_moq_ingest::frame::open(p).unwrap().t;
|
|
let e = spans.entry(*g).or_insert((t, t));
|
|
e.0 = e.0.min(t);
|
|
e.1 = e.1.max(t);
|
|
}
|
|
}
|
|
assert!(spans.values().all(|(a, b)| b - a < 200_000), "every audio group within one window");
|
|
let firsts: Vec<bool> = envelopes(&watcher, video).iter().map(|e| e.key).collect();
|
|
assert_eq!(firsts.iter().filter(|k| **k).count(), 3);
|
|
assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");
|
|
|
|
// The session's record, signed by the gateway.
|
|
assert!(until(STEP, || !w.gateway.gateway.records().is_empty()).await);
|
|
let rec = w.gateway.gateway.records().remove(0);
|
|
rec.verify().expect("the record verifies under the gateway's key");
|
|
assert_eq!((rec.record.protocol.as_str(), rec.record.role, rec.record.ended.as_str()), ("whip", 3, "deleted"));
|
|
assert_eq!(rec.record.energy.acquisition, "unmetered");
|
|
let out: u64 = rec.record.tracks.iter().filter(|t| t.dir == 2).map(|t| t.objects).sum();
|
|
assert_eq!(out, (sent.video.len() + sent.audio.len()) as u64);
|
|
assert_eq!(rec.record.ledger.len(), 1);
|
|
|
|
p.close();
|
|
w.stop().await;
|
|
}
|