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]>
869 lines
38 KiB
Rust
869 lines
38 KiB
Rust
//! A WHEP session (`draft-ietf-wish-whep-05`): a WebRTC player of one owner's tracks of a
|
|
//! room. The gateway subscribes to the tracks and their receipts tracks at the relay,
|
|
//! verifies every object against its receipt, and hands the player the frames that
|
|
//! verify, packetised back to RTP by the WebRTC stack.
|
|
//!
|
|
//! **What a WHEP player cannot see.** WebRTC carries media, not receipts: a WHEP player
|
|
//! receives frames and nothing that proves where they came from. The gateway is the
|
|
//! verifier on its behalf: an object whose receipt does not verify, whose receipt never
|
|
//! comes, or that does not open as an envelope of the track's codec, is dropped and
|
|
//! counted, and never reaches the player. The counts go into the session's signed record.
|
|
//! A WAI-aware player gets the receipts itself by subscribing to the same tracks and their
|
|
//! receipts tracks over MoQ (the relay's WebTransport and WebSocket paths), and verifying
|
|
//! each object as this module does (`extensions/live-ingest.md` §9).
|
|
|
|
use crate::catalog::{self, Declared};
|
|
use crate::frame::{self, Codec};
|
|
use crate::gateway::{check_if_match, forbidden, problem, random_hex, Cmd, Gateway, Handle, Reply, SDP, SDPFRAG};
|
|
use crate::record::{Energy, LedgerEntry, SessionRecord, SignedRecord, TrackEntry, FORMAT};
|
|
use crate::relay::STEP;
|
|
use crate::room::{media, Admitted, Room};
|
|
use crate::rtc;
|
|
use crate::sdp;
|
|
use hyper::StatusCode;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft22::data::FetchEntry;
|
|
use joule_moq::draft22::{ControlMessage, LocationFilter, Parameter, Parameters, StreamHeader, TrackRequest, UniEvent};
|
|
use joule_moq::receipt::{ReceiptVerifier, VerifiedGroup};
|
|
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackName, TrackNamespace};
|
|
use joule_moq_relay::client::Client;
|
|
use joule_moq_relay::wire::{ReqRx, ReqTx};
|
|
use std::collections::{BTreeMap, HashMap, VecDeque};
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
use str0m::change::SdpOffer;
|
|
use str0m::media::{Frequency, MediaKind, MediaTime, Mid};
|
|
use str0m::{Candidate, Event, IceConnectionState, Rtc};
|
|
use tokio::net::UdpSocket;
|
|
use tokio::sync::mpsc;
|
|
|
|
/// Frames held for a player that has not yet connected.
|
|
const HELD_FRAMES: usize = 2048;
|
|
const DISCONNECT_GRACE: Duration = Duration::from_secs(5);
|
|
|
|
/// One object from the relay: on a subgroup stream, by its downstream Track Alias, or on
|
|
/// a fill fetch stream, by the Request ID of the SUBSCRIBE it fills.
|
|
#[derive(Debug)]
|
|
enum In {
|
|
Object { alias: u64, group: u64, object: u64, payload: Vec<u8> },
|
|
Fill { request: u64, group: u64, object: u64, payload: Vec<u8> },
|
|
}
|
|
|
|
/// Read every data stream of `client`, each concurrently and every read bounded, until
|
|
/// the session closes.
|
|
async fn read_streams(client: Arc<Client>, tx: mpsc::UnboundedSender<In>) {
|
|
loop {
|
|
if tx.is_closed() {
|
|
return;
|
|
}
|
|
let Some(mut r) = client.next_stream(Duration::from_secs(1)).await else { continue };
|
|
let tx = tx.clone();
|
|
tokio::spawn(async move {
|
|
let Ok(Ok(h)) = tokio::time::timeout(STEP, r.start()).await else { return };
|
|
loop {
|
|
match tokio::time::timeout(STEP, r.next()).await {
|
|
Ok(Ok(Some(UniEvent::SubgroupObject(o)))) => {
|
|
if let StreamHeader::Subgroup(sh) = &h
|
|
&& !o.payload.is_empty()
|
|
{
|
|
let _ = tx.send(In::Object { alias: sh.track_alias, group: sh.group_id, object: o.object_id, payload: o.payload });
|
|
}
|
|
}
|
|
Ok(Ok(Some(UniEvent::FetchEntry(FetchEntry::Object(o))))) => {
|
|
if let StreamHeader::Fetch(fh) = &h
|
|
&& !o.payload.is_empty()
|
|
{
|
|
let _ = tx.send(In::Fill { request: fh.request_id, group: o.group_id, object: o.object_id, payload: o.payload });
|
|
}
|
|
}
|
|
Ok(Ok(Some(_))) => {}
|
|
_ => return,
|
|
}
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
/// SUBSCRIBE from the start of the track's latest group: the live part from the Next
|
|
/// Object, and a fill of the latest group from its first object (MOQT §3.5, §9.20.9), so a
|
|
/// player starts at a group's first object, a keyframe for video. Returns the Request ID,
|
|
/// the downstream Track Alias, and the request's stream.
|
|
async fn subscribe_joining(client: &Client, ns: TrackNamespace, name: TrackName) -> Result<(u64, u64, ReqTx, ReqRx), String> {
|
|
let fill = Parameters::new().with(Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 1 }));
|
|
let params = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject)).with(Parameter::FillParameters(fill));
|
|
let id = client.next_request_id();
|
|
let (tx, mut rx) = client.request(&ControlMessage::Subscribe(TrackRequest { request_id: id, namespace: ns, name, params })).await?;
|
|
match Client::response(&mut rx).await? {
|
|
ControlMessage::SubscribeOk(ok) => Ok((id, ok.track_alias, tx, rx)),
|
|
ControlMessage::RequestError(e) => Err(format!("REQUEST_ERROR {}: {}", e.code, String::from_utf8_lossy(&e.reason))),
|
|
other => Err(format!("an unexpected {}", other.name())),
|
|
}
|
|
}
|
|
|
|
/// Counts of one track a WHEP session receives.
|
|
#[derive(Clone, Debug, Default, serde::Serialize)]
|
|
pub struct InCounts {
|
|
pub objects_in: u64,
|
|
pub bytes_in: u64,
|
|
pub receipts_in: u64,
|
|
pub groups_verified: u64,
|
|
pub objects_verified: u64,
|
|
/// Refused: the group receipt does not verify under a trusted key.
|
|
pub refused_signature: u64,
|
|
/// Refused: the object does not match its receipt or its proof.
|
|
pub refused_inclusion: u64,
|
|
/// Refused: the receipt names another track, group or object (moq-streaming-format
|
|
/// §4.1, `moqt_mismatch`).
|
|
pub refused_moqt: u64,
|
|
/// Refused: the group receipt does not chain to the group before it.
|
|
pub refused_chain: u64,
|
|
/// Refused: the receipts object does not read as a bundle.
|
|
pub refused_bundle: u64,
|
|
/// The objects of the groups refused above (those counts are of groups).
|
|
pub refused_objects: u64,
|
|
/// Dropped: no receipt came within `hold_ms`.
|
|
pub receipt_missing: u64,
|
|
/// Receipted, not delivered: an object the receipts name that did not arrive.
|
|
pub object_missing: u64,
|
|
/// Dropped: it verified but does not open as an envelope of the track's codec.
|
|
pub unopenable: u64,
|
|
pub late: u64,
|
|
pub duplicate: u64,
|
|
pub frames_out: u64,
|
|
pub bytes_out: u64,
|
|
}
|
|
|
|
struct Pending {
|
|
last_seen: Instant,
|
|
objects: BTreeMap<u64, Vec<u8>>,
|
|
receipts: Option<(ReceiptBundle, VerifiedGroup)>,
|
|
refused: bool,
|
|
next: u64,
|
|
}
|
|
|
|
impl Pending {
|
|
fn new(now: Instant) -> Self {
|
|
Pending { last_seen: now, objects: BTreeMap::new(), receipts: None, refused: false, next: 0 }
|
|
}
|
|
}
|
|
|
|
/// A frame that verified, for the player.
|
|
#[derive(Debug)]
|
|
pub struct Frame {
|
|
pub kind: u8,
|
|
pub codec: Codec,
|
|
pub t: u64,
|
|
pub data: Vec<u8>,
|
|
}
|
|
|
|
/// One track a WHEP session receives, and its verifier.
|
|
pub struct InTrack {
|
|
pub kind: u8,
|
|
pub declared: Declared,
|
|
pub ns: TrackNamespace,
|
|
pub media_alias: u64,
|
|
pub receipts_alias: u64,
|
|
pub media_request: u64,
|
|
pub receipts_request: u64,
|
|
groups: BTreeMap<u64, Pending>,
|
|
links: BTreeMap<u64, [u8; 32]>,
|
|
released: Option<u64>,
|
|
pub counts: InCounts,
|
|
}
|
|
|
|
impl InTrack {
|
|
fn object(&mut self, group: u64, object: u64, payload: Vec<u8>, now: Instant) {
|
|
self.counts.objects_in += 1;
|
|
self.counts.bytes_in += payload.len() as u64;
|
|
if self.released.is_some_and(|r| group <= r) {
|
|
self.counts.late += 1;
|
|
return;
|
|
}
|
|
let p = self.groups.entry(group).or_insert_with(|| Pending::new(now));
|
|
p.last_seen = now;
|
|
if object < p.next || p.objects.insert(object, payload).is_some() {
|
|
self.counts.duplicate += 1;
|
|
}
|
|
}
|
|
|
|
fn receipts(&mut self, group: u64, payload: &[u8], verifier: &ReceiptVerifier, now: Instant) {
|
|
self.counts.receipts_in += 1;
|
|
if self.released.is_some_and(|r| group <= r) {
|
|
return;
|
|
}
|
|
let p = self.groups.entry(group).or_insert_with(|| Pending::new(now));
|
|
if p.receipts.is_some() || p.refused {
|
|
return;
|
|
}
|
|
let Ok(b) = ReceiptBundle::from_payload(payload) else {
|
|
self.counts.refused_bundle += 1;
|
|
p.refused = true;
|
|
return;
|
|
};
|
|
// A receipt names its MOQT group and object, and the track's receipt chain, before
|
|
// its signature is checked (moq-streaming-format §4.1).
|
|
let alias = TrackAlias(self.declared.receipt_alias);
|
|
if b.group.group_id != GroupId(group) || b.group.track_alias != alias || b.objects.iter().enumerate().any(|(i, r)| r.check_moqt(alias, GroupId(group), ObjectId(i as u64)).is_err()) {
|
|
self.counts.refused_moqt += 1;
|
|
p.refused = true;
|
|
return;
|
|
}
|
|
let vg = match verifier.verify_group(&b.group) {
|
|
Ok(v) => v,
|
|
Err(_) => {
|
|
self.counts.refused_signature += 1;
|
|
p.refused = true;
|
|
return;
|
|
}
|
|
};
|
|
if let Some(prev) = group.checked_sub(1).and_then(|g| self.links.get(&g))
|
|
&& b.group.parent_receipt_hash != Some(*prev)
|
|
{
|
|
self.counts.refused_chain += 1;
|
|
p.refused = true;
|
|
return;
|
|
}
|
|
self.links.insert(group, b.group.link_hash());
|
|
self.counts.groups_verified += 1;
|
|
p.receipts = Some((b, vg));
|
|
}
|
|
|
|
/// Hand on, in group order and object order, every object whose receipt verified;
|
|
/// drop and count what fails, or what waited past `hold`.
|
|
fn release(&mut self, verifier: &ReceiptVerifier, hold: Duration, now: Instant, out: &mut Vec<Frame>) {
|
|
let alias = TrackAlias(self.declared.receipt_alias);
|
|
while let Some((&g, p)) = self.groups.iter_mut().next() {
|
|
let expired = now.duration_since(p.last_seen) > hold;
|
|
let done = if p.refused {
|
|
// Its group was refused by its receipt: every object of it is dropped.
|
|
self.counts.refused_objects += p.objects.len() as u64;
|
|
true
|
|
} else if let Some((b, vg)) = &p.receipts {
|
|
let n = b.group.leaf_count;
|
|
while p.next < n {
|
|
let Some(payload) = p.objects.remove(&p.next) else { break };
|
|
let r = &b.objects[p.next as usize];
|
|
let ok = r.check_moqt(alias, GroupId(g), ObjectId(p.next)).is_ok() && verifier.verify_payload(&payload, r, vg).is_ok();
|
|
if !ok {
|
|
self.counts.refused_inclusion += 1;
|
|
} else {
|
|
self.counts.objects_verified += 1;
|
|
match frame::open(&payload).ok().filter(|o| o.codec == self.declared.codec).and_then(|o| frame::unframe(o.codec, &o.payload).map(|d| (o.t, d))) {
|
|
Some((t, data)) => out.push(Frame { kind: self.kind, codec: self.declared.codec, t, data }),
|
|
None => self.counts.unopenable += 1,
|
|
}
|
|
}
|
|
p.next += 1;
|
|
}
|
|
if p.next >= n {
|
|
true
|
|
} else if expired {
|
|
// Receipted and not delivered, where the session held any of the
|
|
// group: a group whose objects all came before it joined is not.
|
|
if p.next > 0 || !p.objects.is_empty() {
|
|
self.counts.object_missing += n - p.next;
|
|
}
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
} else if expired {
|
|
self.counts.receipt_missing += p.objects.len() as u64;
|
|
true
|
|
} else {
|
|
false
|
|
};
|
|
if !done {
|
|
break;
|
|
}
|
|
self.groups.remove(&g);
|
|
self.released = Some(g);
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn create(gw: Arc<Gateway>, room: Arc<Room>, ad: Admitted, token_hash: [u8; 32], owner: String, layer: u8, offer: String) -> Reply {
|
|
if !ad.grant.caps.subscribe {
|
|
return forbidden("subscribe_not_permitted", "the grant does not let its principal subscribe");
|
|
}
|
|
if layer > ad.grant.caps.max_layer {
|
|
return forbidden("layer_not_permitted", "the grant's max_layer is below the layer asked for");
|
|
}
|
|
for s in sdp::sections(&offer) {
|
|
if (s.media == "audio" || s.media == "video") && matches!(s.direction.as_deref(), Some("sendonly") | Some("inactive")) {
|
|
return problem(StatusCode::UNPROCESSABLE_ENTITY, "direction", "a WHEP offer's sections are recvonly or sendrecv (WHEP §4.3.6)");
|
|
}
|
|
}
|
|
if !gw.reserve(true) {
|
|
return gw.busy();
|
|
}
|
|
match open(&gw, &room, &ad, token_hash, &owner, layer, &offer).await {
|
|
Ok(r) => r,
|
|
Err(r) => {
|
|
gw.release(true);
|
|
r
|
|
}
|
|
}
|
|
}
|
|
|
|
fn not_live(gw: &Gateway, detail: &str) -> Reply {
|
|
problem(StatusCode::CONFLICT, "not_live", detail).header(hyper::header::RETRY_AFTER, gw.cfg.whep.retry_after_s.to_string())
|
|
}
|
|
|
|
async fn open(gw: &Arc<Gateway>, room: &Arc<Room>, ad: &Admitted, token_hash: [u8; 32], owner: &str, layer: u8, offer: &str) -> Result<Reply, Reply> {
|
|
let (client, endpoint) = gw.link.connect().await.map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "relay_unreachable", &e))?;
|
|
let client = Arc::new(client);
|
|
let (in_tx, mut in_rx) = mpsc::unbounded_channel();
|
|
tokio::spawn(read_streams(client.clone(), in_tx.clone()));
|
|
let close = |c: &Client| c.close(0, "WHEP session refused");
|
|
|
|
// The catalog: which tracks the owner publishes, in which codec, under which receipt
|
|
// chain (moq-streaming-format §4), read only once its own receipt verifies under a
|
|
// trusted first hop. A player of an owner with none is told to come back (WHEP
|
|
// §4.3.9).
|
|
let verifier = ReceiptVerifier::with_anchors(gw.trust.clone());
|
|
let cat_ns = catalog::catalog_namespace(&room.id, owner);
|
|
let subscribed = async {
|
|
let c = subscribe_joining(&client, cat_ns.clone(), TrackName::from_str(catalog::CATALOG_NAME)).await?;
|
|
let r = subscribe_joining(&client, cat_ns.clone(), catalog::receipts_name(catalog::CATALOG_NAME)).await?;
|
|
Ok::<_, String>((c, r))
|
|
};
|
|
let (cat, cat_r) = match subscribed.await {
|
|
Ok(x) => x,
|
|
Err(e) => {
|
|
close(&client);
|
|
return Err(not_live(gw, &format!("no catalog for this owner: {e}")));
|
|
}
|
|
};
|
|
let deadline = tokio::time::Instant::now() + Duration::from_millis(gw.cfg.whep.catalog_wait_ms);
|
|
let (mut bodies, mut bundles): (BTreeMap<u64, Vec<u8>>, BTreeMap<u64, Vec<u8>>) = (BTreeMap::new(), BTreeMap::new());
|
|
let mut declared = None;
|
|
let mut early = Vec::new();
|
|
while declared.is_none() {
|
|
match tokio::time::timeout_at(deadline, in_rx.recv()).await {
|
|
Ok(Some(In::Object { alias, group, payload, .. })) if alias == cat.1 => drop(bodies.insert(group, payload)),
|
|
Ok(Some(In::Fill { request, group, payload, .. })) if request == cat.0 => drop(bodies.insert(group, payload)),
|
|
Ok(Some(In::Object { alias, group, payload, .. })) if alias == cat_r.1 => drop(bundles.insert(group, payload)),
|
|
Ok(Some(In::Fill { request, group, payload, .. })) if request == cat_r.0 => drop(bundles.insert(group, payload)),
|
|
Ok(Some(other)) => early.push(other),
|
|
_ => break,
|
|
}
|
|
// The latest group whose catalog and receipt are both held, and verify.
|
|
for (g, body) in bodies.iter().rev() {
|
|
if let Some(b) = bundles.get(g).and_then(|b| ReceiptBundle::from_payload(b).ok())
|
|
&& b.group.group_id == GroupId(*g)
|
|
&& b.objects.len() == 1
|
|
&& b.objects[0].group_id == GroupId(*g)
|
|
&& b.objects[0].object_id == ObjectId(0)
|
|
&& let Ok(vg) = verifier.verify_group(&b.group)
|
|
&& verifier.verify_payload(body, &b.objects[0], &vg).is_ok()
|
|
{
|
|
declared = catalog::read(body);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
let Some(declared) = declared else {
|
|
close(&client);
|
|
return Err(not_live(gw, "no catalog of the owner's, with a receipt that verifies, arrived"));
|
|
};
|
|
let mut want: Vec<(u8, Declared)> = Vec::new();
|
|
for m in [media::AUDIO, media::VIDEO] {
|
|
let ns = catalog::render(&catalog::media_namespace(&room.id, owner, m));
|
|
if let Some(d) = declared.iter().find(|d| d.namespace == ns && d.name == layer.to_string() && d.codec.is_video() == (m == media::VIDEO)) {
|
|
want.push((m, d.clone()));
|
|
}
|
|
}
|
|
if want.is_empty() {
|
|
close(&client);
|
|
return Err(problem(StatusCode::NOT_FOUND, "layer_unknown", "the owner's catalog declares no track of this layer"));
|
|
}
|
|
|
|
// The answer: only the codecs the tracks carry, so a player that offers none of a
|
|
// kind's has that section rejected (WHEP §4.3.4).
|
|
let socket = UdpSocket::bind((gw.cfg.media.bind_ip, 0)).await.map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "media_socket", &e.to_string()))?;
|
|
let local = socket.local_addr().expect("a bound socket");
|
|
let codecs: Vec<Codec> = want.iter().map(|(_, d)| d.codec).collect();
|
|
let mut rtc = rtc::config(gw.cfg.media.ice_lite, &codecs).build(Instant::now());
|
|
let cand = Candidate::host(local, "udp").map_err(|e| problem(StatusCode::SERVICE_UNAVAILABLE, "media_candidate", &e.to_string()))?;
|
|
rtc.add_local_candidate(cand);
|
|
let parsed = SdpOffer::from_sdp_string(offer).map_err(|e| problem(StatusCode::BAD_REQUEST, "offer_malformed", &e.to_string()))?;
|
|
let answer = rtc.sdp_api().accept_offer(parsed).map_err(|e| problem(StatusCode::UNPROCESSABLE_ENTITY, "offer_refused", &e.to_string()))?.to_sdp_string();
|
|
let accepted: Vec<u8> = sdp::sections(&answer)
|
|
.iter()
|
|
.filter(|s| s.port != 0)
|
|
.filter_map(|s| match s.media.as_str() {
|
|
"audio" => Some(media::AUDIO),
|
|
"video" => Some(media::VIDEO),
|
|
_ => None,
|
|
})
|
|
.collect();
|
|
let mut tracks = Vec::new();
|
|
let mut done_watch = Vec::new();
|
|
for (m, d) in want.into_iter().filter(|(m, _)| accepted.contains(m)) {
|
|
let ns = catalog::media_namespace(&room.id, owner, m);
|
|
let (mr, ma, mtx, mrx) = subscribe_joining(&client, ns.clone(), TrackName::from_str(&d.name)).await.map_err(|e| not_live(gw, &e))?;
|
|
let (rr, ra, rtx, rrx) = subscribe_joining(&client, ns.clone(), TrackName::from_str(&d.receipts_track)).await.map_err(|e| not_live(gw, &e))?;
|
|
done_watch.push((mtx, mrx));
|
|
done_watch.push((rtx, rrx));
|
|
tracks.push(InTrack {
|
|
kind: m,
|
|
declared: d,
|
|
ns,
|
|
media_alias: ma,
|
|
receipts_alias: ra,
|
|
media_request: mr,
|
|
receipts_request: rr,
|
|
groups: BTreeMap::new(),
|
|
links: BTreeMap::new(),
|
|
released: None,
|
|
counts: InCounts::default(),
|
|
});
|
|
}
|
|
if tracks.is_empty() {
|
|
close(&client);
|
|
return Err(problem(StatusCode::UNPROCESSABLE_ENTITY, "no_common_codec", "the offer names none of the codecs the owner's tracks carry"));
|
|
}
|
|
// A track's end at the relay (PUBLISH_DONE, MOQT §9.9) ends the session.
|
|
let (end_tx, end_rx) = mpsc::unbounded_channel::<String>();
|
|
for (i, (_tx, mut rx)) in done_watch.into_iter().enumerate() {
|
|
let end_tx = end_tx.clone();
|
|
tokio::spawn(async move {
|
|
let _keep = _tx;
|
|
loop {
|
|
match rx.next().await {
|
|
Ok(Some(ControlMessage::PublishDone(d))) => {
|
|
if i % 2 == 0 {
|
|
let _ = end_tx.send(format!("track_ended ({:?})", d.status));
|
|
}
|
|
return;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
_ => return,
|
|
}
|
|
}
|
|
});
|
|
}
|
|
for e in early {
|
|
let _ = in_tx.send(e);
|
|
}
|
|
drop(in_tx);
|
|
|
|
let id = random_hex(16);
|
|
let etag = random_hex(16);
|
|
let (tx, rx) = mpsc::channel(16);
|
|
let frames = Arc::new(std::sync::atomic::AtomicU64::new(0));
|
|
gw.whep.lock().unwrap().insert(id.clone(), Handle { tx, token_hash, frames: frames.clone() });
|
|
gw.stats.whep_opened.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
|
let s = Session {
|
|
gw: gw.clone(),
|
|
id: id.clone(),
|
|
room: room.clone(),
|
|
ad: ad.clone(),
|
|
layer,
|
|
rtc,
|
|
offer: offer.to_string(),
|
|
etag: etag.clone(),
|
|
client,
|
|
_endpoint: endpoint,
|
|
verifier,
|
|
tracks,
|
|
mids: HashMap::new(),
|
|
held: VecDeque::new(),
|
|
held_dropped: 0,
|
|
base: None,
|
|
connected: false,
|
|
created: Instant::now(),
|
|
disconnected_since: None,
|
|
start_us: 0,
|
|
frames,
|
|
};
|
|
tokio::spawn(s.run(socket, local, rx, in_rx, end_rx));
|
|
Ok(Reply::new(StatusCode::CREATED).header(hyper::header::LOCATION, format!("/whep/s/{id}")).header(hyper::header::ETAG, format!("\"{etag}\"")).body(SDP, answer))
|
|
}
|
|
|
|
struct Session {
|
|
gw: Arc<Gateway>,
|
|
id: String,
|
|
room: Arc<Room>,
|
|
ad: Admitted,
|
|
layer: u8,
|
|
rtc: Rtc,
|
|
offer: String,
|
|
etag: String,
|
|
client: Arc<Client>,
|
|
_endpoint: quinn::Endpoint,
|
|
verifier: ReceiptVerifier,
|
|
tracks: Vec<InTrack>,
|
|
mids: HashMap<u8, Mid>,
|
|
held: VecDeque<Frame>,
|
|
held_dropped: u64,
|
|
/// The instant and room-clock time of the first frame written: every frame's
|
|
/// wallclock is placed from it, so its tracks keep their relative timing.
|
|
base: Option<(Instant, u64)>,
|
|
connected: bool,
|
|
created: Instant,
|
|
disconnected_since: Option<Instant>,
|
|
start_us: u64,
|
|
frames: Arc<std::sync::atomic::AtomicU64>,
|
|
}
|
|
|
|
impl Session {
|
|
fn room_now(&self) -> u64 {
|
|
self.gw.clock.unix_us().saturating_sub(self.room.descriptor.t0_us)
|
|
}
|
|
|
|
async fn run(mut self, socket: UdpSocket, local: std::net::SocketAddr, mut rx: mpsc::Receiver<Cmd>, mut in_rx: mpsc::UnboundedReceiver<In>, mut end_rx: mpsc::UnboundedReceiver<String>) {
|
|
self.start_us = self.room_now();
|
|
let mut buf = vec![0u8; 2048];
|
|
let mut events = Vec::new();
|
|
let mut tick = tokio::time::interval(Duration::from_millis(50));
|
|
let mut delete_reply = None;
|
|
let hold = Duration::from_millis(self.gw.cfg.whep.hold_ms);
|
|
let ended: String = 'run: loop {
|
|
let next = match rtc::drain(&mut self.rtc, &socket, &mut events) {
|
|
Ok(t) => t,
|
|
Err(e) => break 'run format!("webrtc: {e}"),
|
|
};
|
|
if !events.is_empty() {
|
|
let evs: Vec<Event> = std::mem::take(&mut events);
|
|
for e in evs {
|
|
self.on_event(e);
|
|
}
|
|
continue;
|
|
}
|
|
let sleep = tokio::time::sleep_until(tokio::time::Instant::from_std(next.max(Instant::now())));
|
|
tokio::select! {
|
|
r = socket.recv_from(&mut buf) => {
|
|
if let Ok((n, from)) = r
|
|
&& let Err(e) = rtc::input(&mut self.rtc, local, from, &buf[..n])
|
|
{
|
|
break 'run format!("webrtc: {e}");
|
|
}
|
|
}
|
|
_ = sleep => {
|
|
if let Err(e) = self.rtc.handle_input(str0m::Input::Timeout(Instant::now())) {
|
|
break 'run format!("webrtc: {e}");
|
|
}
|
|
}
|
|
i = in_rx.recv() => {
|
|
let Some(i) = i else { break 'run "relay_lost".into() };
|
|
self.on_in(i);
|
|
while let Ok(i) = in_rx.try_recv() {
|
|
self.on_in(i);
|
|
}
|
|
self.release(hold);
|
|
}
|
|
why = end_rx.recv() => {
|
|
// Whatever verified is handed on before the session ends.
|
|
while let Ok(i) = in_rx.try_recv() {
|
|
self.on_in(i);
|
|
}
|
|
self.release(Duration::ZERO);
|
|
break 'run why.unwrap_or_else(|| "track_ended".into());
|
|
}
|
|
c = rx.recv() => match c {
|
|
None => break 'run "gateway_stopped".into(),
|
|
Some(Cmd::Delete { reply }) => {
|
|
delete_reply = Some(reply);
|
|
break 'run "deleted".into();
|
|
}
|
|
Some(Cmd::Patch { frag, if_match, reply }) => {
|
|
let r = self.patch(frag, if_match.as_deref());
|
|
let _ = reply.send(r);
|
|
}
|
|
},
|
|
_ = tick.tick() => {
|
|
self.release(hold);
|
|
let m = &self.gw.cfg.media;
|
|
if !self.connected && self.created.elapsed() > Duration::from_millis(m.connect_timeout_ms) {
|
|
break 'run "connect_timeout".into();
|
|
}
|
|
if self.disconnected_since.is_some_and(|t| t.elapsed() > DISCONNECT_GRACE) {
|
|
break 'run "ice_disconnected".into();
|
|
}
|
|
}
|
|
}
|
|
};
|
|
self.end(&ended).await;
|
|
if let Some(r) = delete_reply {
|
|
let _ = r.send(());
|
|
}
|
|
}
|
|
|
|
fn on_event(&mut self, e: Event) {
|
|
match e {
|
|
Event::Connected => {
|
|
self.connected = true;
|
|
while let Some(f) = self.held.pop_front() {
|
|
self.write(f);
|
|
}
|
|
}
|
|
Event::IceConnectionStateChange(s) => match s {
|
|
IceConnectionState::Disconnected => {
|
|
self.disconnected_since.get_or_insert_with(Instant::now);
|
|
}
|
|
IceConnectionState::Connected | IceConnectionState::Completed => self.disconnected_since = None,
|
|
_ => {}
|
|
},
|
|
Event::MediaAdded(m) => {
|
|
let k = if m.kind == MediaKind::Audio { media::AUDIO } else { media::VIDEO };
|
|
self.mids.insert(k, m.mid);
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
fn on_in(&mut self, i: In) {
|
|
let now = Instant::now();
|
|
for t in self.tracks.iter_mut() {
|
|
match &i {
|
|
In::Object { alias, group, object, payload } if *alias == t.media_alias => return t.object(*group, *object, payload.clone(), now),
|
|
In::Fill { request, group, object, payload } if *request == t.media_request => return t.object(*group, *object, payload.clone(), now),
|
|
In::Object { alias, group, payload, .. } if *alias == t.receipts_alias => return t.receipts(*group, payload, &self.verifier, now),
|
|
In::Fill { request, group, payload, .. } if *request == t.receipts_request => return t.receipts(*group, payload, &self.verifier, now),
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn release(&mut self, hold: Duration) {
|
|
let now = Instant::now();
|
|
let mut out = Vec::new();
|
|
for t in self.tracks.iter_mut() {
|
|
t.release(&self.verifier, hold, now, &mut out);
|
|
}
|
|
for f in out {
|
|
if self.connected {
|
|
self.write(f);
|
|
} else {
|
|
if self.held.len() >= HELD_FRAMES {
|
|
self.held.pop_front();
|
|
self.held_dropped += 1;
|
|
}
|
|
self.held.push_back(f);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// One verified frame to the player: its RTP time from its room-clock time, its
|
|
/// wallclock placed from the session's first frame.
|
|
fn write(&mut self, f: Frame) {
|
|
let Some(&mid) = self.mids.get(&f.kind) else { return };
|
|
let Some(pt) = self.rtc.writer(mid).and_then(|w| w.payload_params().find(|p| p.spec().codec == f.codec.to_str0m()).map(|p| p.pt())) else { return };
|
|
let (base_i, base_t) = *self.base.get_or_insert((Instant::now(), f.t));
|
|
let wallclock = if f.t >= base_t { base_i + Duration::from_micros(f.t - base_t) } else { base_i.checked_sub(Duration::from_micros(base_t - f.t)).unwrap_or(base_i) };
|
|
let (ticks, freq) = if f.codec.is_video() { (f.t as u128 * 90_000 / 1_000_000, Frequency::NINETY_KHZ) } else { (f.t as u128 * 48_000 / 1_000_000, Frequency::FORTY_EIGHT_KHZ) };
|
|
let n = f.data.len() as u64;
|
|
if let Some(w) = self.rtc.writer(mid)
|
|
&& w.write(pt, wallclock, MediaTime::new(ticks as u64, freq), f.data).is_ok()
|
|
&& let Some(t) = self.tracks.iter_mut().find(|t| t.kind == f.kind)
|
|
{
|
|
t.counts.frames_out += 1;
|
|
t.counts.bytes_out += n;
|
|
self.frames.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
|
}
|
|
// A write changes the session: the caller's loop drains it before it waits.
|
|
}
|
|
|
|
fn patch(&mut self, frag: sdp::Frag, if_match: Option<&str>) -> Reply {
|
|
if let Err(r) = check_if_match(if_match, &self.etag) {
|
|
return r;
|
|
}
|
|
let current = (sdp::attr(&self.offer, "ice-ufrag").map(str::to_string), sdp::attr(&self.offer, "ice-pwd").map(str::to_string));
|
|
let restart = matches!((&frag.ufrag, &frag.pwd), (Some(u), Some(p)) if (Some(u.clone()), Some(p.clone())) != current);
|
|
if restart {
|
|
let (u, p) = (frag.ufrag.as_deref().unwrap(), frag.pwd.as_deref().unwrap());
|
|
let offer = sdp::restart_offer(&self.offer, u, p, &frag.candidates, frag.end_of_candidates);
|
|
let parsed = match SdpOffer::from_sdp_string(&offer) {
|
|
Ok(o) => o,
|
|
Err(e) => return problem(StatusCode::BAD_REQUEST, "sdpfrag_malformed", &e.to_string()),
|
|
};
|
|
match self.rtc.sdp_api().accept_offer(parsed) {
|
|
Ok(a) => {
|
|
self.offer = offer;
|
|
self.etag = random_hex(16);
|
|
Reply::new(StatusCode::OK).header(hyper::header::ETAG, format!("\"{}\"", self.etag)).body(SDPFRAG, sdp::restart_frag(&a.to_sdp_string()))
|
|
}
|
|
Err(e) => problem(StatusCode::UNPROCESSABLE_ENTITY, "ice_restart_refused", &e.to_string()),
|
|
}
|
|
} else {
|
|
for c in &frag.candidates {
|
|
if let Ok(c) = Candidate::from_sdp_string(c) {
|
|
self.rtc.add_remote_candidate(c);
|
|
}
|
|
}
|
|
Reply::new(StatusCode::NO_CONTENT)
|
|
}
|
|
}
|
|
|
|
async fn end(mut self, why: &str) {
|
|
self.client.close(0, "the WHEP session ended");
|
|
self.rtc.disconnect();
|
|
let end_us = self.room_now();
|
|
let record = self.record(why, end_us);
|
|
self.gw.sink.record(&SignedRecord::sign(record, &self.gw.signer.key));
|
|
self.gw.forget(true, &self.id);
|
|
self.gw.release(true);
|
|
tracing::info!(session = %self.id, "WHEP session ended: {why}");
|
|
}
|
|
|
|
fn record(&self, why: &str, end: u64) -> SessionRecord {
|
|
let mut tracks = Vec::new();
|
|
let mut detail = serde_json::Map::new();
|
|
let (mut bytes_in, mut bytes_out) = (0, 0);
|
|
for t in &self.tracks {
|
|
let digest = hex::encode(catalog::track_digest(&t.ns, &t.declared.name));
|
|
let base = TrackEntry { kind: 1, track: digest, namespace: catalog::render(&t.ns), name: t.declared.name.clone(), ..Default::default() };
|
|
let c = &t.counts;
|
|
let refused = c.refused_objects + c.refused_inclusion + c.receipt_missing + c.unopenable;
|
|
tracks.push(TrackEntry { dir: 1, objects: c.objects_in, bytes: c.bytes_in, dropped_refused: refused, dropped_duplicate: c.duplicate, ..base.clone() });
|
|
tracks.push(TrackEntry { dir: 2, objects: c.frames_out, bytes: c.bytes_out, receipted: c.object_missing, ..base });
|
|
bytes_in += c.bytes_in;
|
|
bytes_out += c.bytes_out;
|
|
detail.insert(crate::room::media_name(t.kind).unwrap_or("media").into(), serde_json::to_value(c).unwrap());
|
|
}
|
|
tracks.sort_by(|a, b| (a.kind, &a.track, a.dir).cmp(&(b.kind, &b.track, b.dir)));
|
|
detail.insert("held_dropped".into(), self.held_dropped.into());
|
|
let duration = end.saturating_sub(self.start_us);
|
|
let mut subscriber_us = [0u64; 9];
|
|
subscriber_us[(self.layer as usize).min(7)] = duration;
|
|
let g = &self.ad.grant;
|
|
SessionRecord {
|
|
format: FORMAT.into(),
|
|
hop: hex::encode(self.gw.public_key()),
|
|
role: 3,
|
|
protocol: "whep".into(),
|
|
session: self.id.clone(),
|
|
room: hex::encode(self.room.id),
|
|
principal: hex::encode(g.principal),
|
|
start: self.start_us,
|
|
end: end.max(self.start_us + 1),
|
|
energy: Energy::unmetered(),
|
|
tracks,
|
|
sessions: 1,
|
|
subscriber_us,
|
|
ledger: vec![LedgerEntry {
|
|
grant: hex::encode(self.ad.digest),
|
|
joules_ceiling: g.joules_ceiling,
|
|
funds_ceiling: g.funds_ceiling,
|
|
currency: String::from_utf8_lossy(&g.currency).trim_matches('\0').to_string(),
|
|
sessions: 1,
|
|
bytes_in,
|
|
bytes_out,
|
|
subscriber_us,
|
|
}],
|
|
ended: why.into(),
|
|
detail: serde_json::Value::Object(detail),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::frame::GrantRef;
|
|
use crate::package::{Emit, Packager, Signer, TrackOut};
|
|
use ed25519_dalek::SigningKey;
|
|
|
|
fn declared(alias: u64) -> Declared {
|
|
Declared { namespace: "n".into(), name: "0".into(), codec: Codec::Opus, receipt_alias: alias, receipts_track: "0_receipts".into() }
|
|
}
|
|
|
|
fn track(alias: u64) -> InTrack {
|
|
InTrack {
|
|
kind: media::AUDIO,
|
|
declared: declared(alias),
|
|
ns: TrackNamespace::from_strs(["n"]),
|
|
media_alias: 2,
|
|
receipts_alias: 3,
|
|
media_request: 10,
|
|
receipts_request: 12,
|
|
groups: BTreeMap::new(),
|
|
links: BTreeMap::new(),
|
|
released: None,
|
|
counts: InCounts::default(),
|
|
}
|
|
}
|
|
|
|
/// Two audio groups as a WHIP session packages and signs them.
|
|
fn signed(key: &SigningKey, alias: u64) -> Vec<Emit> {
|
|
let signer = Signer { key: Arc::new(key.clone()), signer_id: "x".into(), origin_label: "wai-ingest".into() };
|
|
let out = TrackOut { media: media::AUDIO, media_alias: 2, receipts_alias: 3, receipt_alias: alias, priority: 0 };
|
|
let mut p = Packager::new(out, 48_000, 250_000, GrantRef { grant_hash: [1; 32], joule_ceiling_micro: u64::MAX, funds_ceiling: None }, signer, 0, 200_000, 2_000_000, 1 << 20);
|
|
let mut e = Vec::new();
|
|
for i in 0..20u64 {
|
|
p.on_frame(Codec::Opus, &[0xfc, i as u8, 7], 960 * i, 1_000_000 + 20_000 * i, false, &mut e);
|
|
}
|
|
p.end_group(&mut e);
|
|
e
|
|
}
|
|
|
|
fn feed(t: &mut InTrack, emits: &[Emit], v: &ReceiptVerifier, tamper: Option<(u64, u64)>) -> Vec<Frame> {
|
|
let now = Instant::now();
|
|
for e in emits {
|
|
match e {
|
|
Emit::Object { group, object, bytes, .. } => {
|
|
let mut b = bytes.clone();
|
|
if tamper == Some((*group, *object)) {
|
|
*b.last_mut().unwrap() ^= 1;
|
|
}
|
|
t.object(*group, *object, b, now);
|
|
}
|
|
Emit::Single { group, bytes, .. } => t.receipts(*group, bytes, v, now),
|
|
Emit::EndGroup { .. } => {}
|
|
}
|
|
}
|
|
let mut out = Vec::new();
|
|
t.release(v, Duration::from_secs(5), now, &mut out);
|
|
out
|
|
}
|
|
|
|
#[test]
|
|
fn every_object_is_verified_before_it_is_handed_on() {
|
|
let key = SigningKey::from_bytes(&[4; 32]);
|
|
let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
let e = signed(&key, 41);
|
|
let mut t = track(41);
|
|
let out = feed(&mut t, &e, &v, None);
|
|
assert_eq!(out.len(), 20);
|
|
assert!(out.iter().enumerate().all(|(i, f)| f.data == vec![0xfc, i as u8, 7]));
|
|
assert_eq!(t.counts.objects_verified, 20);
|
|
assert_eq!(t.counts.groups_verified, 2);
|
|
}
|
|
|
|
#[test]
|
|
fn what_fails_is_dropped_and_counted() {
|
|
let key = SigningKey::from_bytes(&[4; 32]);
|
|
let e = signed(&key, 41);
|
|
// A changed object.
|
|
let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
let first = e.iter().find_map(|x| if let Emit::Object { group, .. } = x { Some(*group) } else { None }).unwrap();
|
|
let mut t = track(41);
|
|
let out = feed(&mut t, &e, &v, Some((first, 3)));
|
|
assert_eq!((out.len(), t.counts.refused_inclusion), (19, 1));
|
|
// An untrusted signer: every object of both groups refused.
|
|
let other = ReceiptVerifier::with_anchors(vec![SigningKey::from_bytes(&[5; 32]).verifying_key()]);
|
|
let mut t = track(41);
|
|
assert!(feed(&mut t, &e, &other, None).is_empty());
|
|
assert_eq!(t.counts.refused_signature, 2);
|
|
// Receipts for another chain.
|
|
let mut t = track(42);
|
|
assert!(feed(&mut t, &e, &v, None).is_empty());
|
|
assert_eq!(t.counts.refused_moqt, 2);
|
|
// No receipts at all: held, then dropped past the hold.
|
|
let mut t = track(41);
|
|
let now = Instant::now();
|
|
for x in &e {
|
|
if let Emit::Object { group, object, bytes, .. } = x {
|
|
t.object(*group, *object, bytes.clone(), now);
|
|
}
|
|
}
|
|
let mut out = Vec::new();
|
|
t.release(&v, Duration::from_secs(5), now, &mut out);
|
|
assert!(out.is_empty());
|
|
t.release(&v, Duration::from_millis(0), now + Duration::from_millis(10), &mut out);
|
|
assert!(out.is_empty());
|
|
assert_eq!(t.counts.receipt_missing, 20);
|
|
}
|
|
}
|