open-standards/joule-code/joule-code-rs/crates/joule-moq-web/src/lib.rs
Transaction Science 6250fa5a4e
Some checks failed
joule-moq-remote host agent / joule-moq-remote · does this change touch it (push) Successful in 33s
WAI operation catalogue / does this change touch it (push) Successful in 28s
wai-app and Create / wai-app · does this change touch it (push) Successful in 29s
joule-moq-room room service / joule-moq-room · does this change touch it (push) Successful in 30s
WAI Studio / does this change touch it (push) Successful in 16s
wai-sink-swift native sink package / does this change touch it (push) Successful in 1m32s
wai-quantum studio and desktop / does this change touch it (push) Successful in 1m34s
wai-fleet service / does this change touch it (push) Successful in 1m40s
wai-audit collector and player integration / does this change touch it (push) Successful in 1m39s
wai-ledger service / does this change touch it (push) Successful in 1m37s
wai-quantum for JavaScript / does this change touch it (push) Successful in 1m34s
joule-moq-web browser client / joule-moq-web · does this change touch it (push) Successful in 1m41s
wai-sink-android native sink library / does this change touch it (push) Successful in 1m35s
joule-moq-remote host agent / joule-moq-remote, every target (linux-arm64) (push) Successful in 2m11s
repository checks / lockfiles, advisories, tests in CI, workflow hygiene, spec and section references, public text (push) Failing after 4m29s
joule-code conformance gate / joule-moq-relay, every target (push) Successful in 7m32s
joule-code conformance gate / joule-moq-ingest, every target (push) Successful in 9m9s
wai-sink-android native sink library / the JNI layer and the platform-free half on the host JVM · linux-arm64 (push) Successful in 6m1s
joule-moq-room room service / joule-moq-room, every target (push) Successful in 8m1s
WAI breaking changes and the conformance runner / WAI breaking · does this change touch WAI (push) Successful in 11s
wai-app and Create / wai-app · tests, clippy, the line language's verifier (linux) (push) Successful in 45s
WAI Studio / the catalogue's actions, the unit tests and the core · linux (push) Successful in 1m22s
joule-code conformance gate / Build + round-trip C1-C4 vectors (push) Successful in 2m54s
WAI operation catalogue / the catalogue and every door · linux (push) Failing after 2m27s
wai-audit collector and player integration / collector · linux (push) Successful in 2m23s
wai-ledger service / service · linux (push) Successful in 3m55s
wai-fleet service / service · linux (push) Successful in 4m31s
wai-quantum for JavaScript / javascript · linux (push) Successful in 6m17s
wai-quantum studio and desktop / studio · browser (push) Successful in 7m43s
wai-quantum studio and desktop / desktop app · linux (push) Successful in 10m31s
WAI Studio / Studio in WebKit · linux (push) Successful in 22m31s
WAI breaking changes and the conformance runner / WAI breaking and conform · their tests, and the independent verifiers through wai conform · linux (push) Failing after 1m31s
WAI breaking changes and the conformance runner / WAI breaking · wire, record and API against main · linux (push) Successful in 3m35s
wai release binaries build on Windows / the seven release binaries · windows (push) Successful in 18m5s
wai-quantum studio and desktop / desktop app · windows (push) Successful in 10m46s
joule-code conformance gate / joule-moq-relay, its own energy on a counter (macos) (push) Successful in 1m3s
joule-moq-remote host agent / joule-moq-remote, every target (macos) (push) Successful in 56s
wai-app and Create / wai-app · Create, browser end-to-end (macos) (push) Successful in 7m44s
WAI Studio / Studio in Chromium and Firefox · macos (push) Successful in 9m17s
wai-fleet service / service · macos (push) Successful in 1m4s
wai-quantum studio and desktop / desktop app · macos (push) Successful in 4m41s
wai-ledger service / service · macos (push) Successful in 1m7s
wai-audit collector and player integration / collector, player integration and the browser end to end · macos (push) Successful in 2m35s
wai-sink-android native sink library / every ABI, the AAR, and the instrumented tests in a headless emulator · macos (push) Successful in 7m17s
wai-sink-swift native sink package / libwai slices, desktop tests, simulator tests · macos (push) Successful in 15m0s
joule-moq-web browser client / joule-moq-web · browser end-to-end (macos) (push) Successful in 15m50s
The relay reads the client's SETUP first (Privacy Pass auth -04), and what the draft-23 package left
SETUP order. draft-ietf-moq-privacy-pass-auth-04 §3.5.1 has a relay that
authenticates SETUP process the client's SETUP before it sends its own; WAI
age-assurance §5.1 and live-rooms §10.5.2 had the relay send first. The decision
was to follow -04, and the texts, the relay, the clients and the corpora do.

The relay's SETUP used to carry the challenge the client's SETUP then proved its
grant over, which the new order cannot do in one session. It is now -04's own
flow (§3.5.6.1, Figure 4), and live-rooms §10.5.3's over HTTP: every SETUP a
relay sends carries a fresh GRANT_CHALLENGE, good for the proofs of one later
SETUP at that relay for limits.grant_challenge_ms (60 s), kept by the relay
(src/challenge.rs) and consumed by the SETUP it admits. A client that holds none
presents its grant unproved, is refused pop_missing with a challenge, and proves
on a second session; one that holds the challenge of the relay's last SETUP to
it proves on its first. A refused SETUP is answered with the relay's SETUP
carrying a new Setup Option, GRANT_REFUSED (0x3A53, the reason's code), and the
session is closed UNAUTHORIZED once the client has left or
limits.setup_refusal_wait_ms (2 s) has passed: a close carries nothing sent
before it, which -04 notes of its own challenge, so the refusal is said in the
SETUP and the client that reads it leaves at once. A consumed challenge is
pop_replayed; one never issued, another relay's, or too old, pop_wrong_challenge.

The clients: joule_moq_relay::client (Client::open, connect_proving, challenge;
its SETUP always first), the room service's clock and participants, the remote
host agent, relay-to-relay links (a link's first SETUP is refused
relay_id_unproved and it dials again at once with its proof), and the browser
(MoqSession.connect's redial, SetupRefused). The interop client and the ingest
gateway already sent first and present no grant. The order is the same at every
draft; grants are proved at -23, -22, -18 and -16 in the relay's test. Not
interoperable with the order before where a grant is presented: a client that
waits for the relay's SETUP is sent none, and relays that peer change together.
Sessions that present no grant are unaffected.

What the draft-23 package left:
- A -23 session's subgroup streams and datagrams are read by -23's object rules
  (a supported property outside its scope or repeated makes the track malformed;
  a grease value in the Mandatory Track Property range is no Mandatory Track
  Property); a -22 session keeps -22's reading.
- Relays peer at -23: RELAY_PATH in the -23 codec and its bridge, a relay id
  taken at -23, links offering -23 and -22. An unnegotiated path closes a -23
  session INVALID_PARAMETER. The -23 corpus gains 6 vectors (125, 70 negative)
  and moqt23_check.py reads the extension.
- A grant that expires ends its subscriptions, each with PUBLISH_DONE
  EXPIRED_AUTH_TOKEN, and not the session, at -23; a -23 session that publishes
  by the grant, and every session at the older drafts, is closed as before.
- LARGEST_OBJECT against -23 §7.5: confirmed but for one source, a publisher's
  PUBLISH_STATE_NOTIFY, which is taken now; held by a vector of six steps. Kept
  and stated: an object counts once whole, not from its first byte.

Tests, each against planted faults: grant_proof (the relay sends nothing before
the client's SETUP; the second session; seven refusals, each with its reason),
room_authority (expiry at -23 beside -22), draft23 (the object rules against the
same objects at -22; the LARGEST_OBJECT vector), cascade and control_sessions
(a link at -23, opened on its second dial; the path at -23), the browser
session's unit tests. Falsified: twelve of the changes reverted in turn, each
failing a test.

live-rooms §10.5.2, §10.5.4, §10.7 and §13, and age-assurance §5.1 and §5.5,
say the order and what a challenge is bound to now. The rooms corpus renames one
proof vector (acknowledged in the change note); the registry lists
GRANT_REFUSED.

Co-Authored-By: Claude <[email protected]>
2026-11-10 17:22:02 -05:00

1240 lines
58 KiB
Rust

//! **joule-moq-web**: `joule-moq`'s MOQT codec (`draft-ietf-moq-transport-22`, and `-23`
//! through `joule-moq`'s bridge) and its receipts, for a browser.
//!
//! The page speaks the draft its transport negotiated: [`set_draft`] (`setDraft`, 22 or 23)
//! sets it once the session is open, and every encoder and every decoder made after it
//! speaks that draft on the wire. The page itself is written in -22's model: a -23 session's
//! messages and fetch entries are mapped to and from -22's by `joule_moq::draft23::bridge`,
//! so the JSON a page reads and writes is -22's view whichever draft is on the wire.
//!
//! The crate builds `joule-moq` without its async runtime (`default-features = false`), so
//! what runs in the page is the same codec and the same receipt verifier the relay runs,
//! compiled for `wasm32-unknown-unknown`. The surface is small and byte-oriented: a page
//! hands it the bytes it read from a stream and gets back the messages and objects in
//! them, and asks it for the bytes of what it sends.
//!
//! - Control: the client's SETUP ([`client_setup`]); every request and answer a client
//! sends, from a JSON description ([`encode_control`]); framed control messages read
//! from a control or request stream as their bytes arrive ([`MessageStream`]).
//! - Data: subgroup streams written object by object ([`SubgroupWriter`]); subgroup and
//! fetch streams read as their bytes arrive ([`UniDecoder`]).
//! - Receipts: a publisher signs each group of a media track into its group receipt and
//! writes the group's receipts-track object and the media object with its `WAI_RECEIPT`
//! property ([`TrackSigner`]); a subscriber admits each receipts-track object to its
//! chain and verifies each media object against it, or accounts for an object it did
//! not receive by inclusion ([`TrackVerifier`]).
//!
//! Signatures are Ed25519, made and verified here by `ed25519-dalek` under the rules
//! `joule-moq`'s verifier holds every receipt to. WebCrypto's Ed25519 is not used: the
//! receipt verifier also needs BLAKE3 and the Merkle walk, which WebCrypto does not have,
//! and one implementation keeps the page's verdicts the relay's.
//!
//! JSON crosses the boundary as strings, in the view `joule-moq`'s vector corpus uses
//! (`draft22::json`): byte strings in lowercase hexadecimal, a namespace as a list of its
//! fields in hexadecimal. Object payloads cross as bytes.
use ed25519_dalek::SigningKey;
use joule_moq::carriage::{object_properties, ObjectExtras, ReceiptBundle, WaiReceiptProperty, SETUP_OPTION_JWP_RECEIPTS};
use joule_moq::draft22::data::{FetchEntry, Limits, ObjectStatus as WireStatus, SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject, UniStreamDecoder};
use joule_moq::draft22::kvp::{opt, prop, KeyValuePair, Properties, SetupOptions};
use joule_moq::draft22::params::{GroupOrder, LocationFilter, Parameter, Parameters};
use joule_moq::draft22::{
json, ControlMessage, Location, NamespaceRequest, Publish, PublishDone, PublishDoneCode, RequestError, RequestErrorCode, RequestOk, RequestUpdate, Setup,
StreamHeader, SubscribeOk, Token, TrackRequest, UniEvent, WireError,
};
use joule_moq::grant_proof::{self, GrantProof};
use joule_moq::messages::ObjectHeader;
use joule_moq::receipt::{Admission, ChainRefusal, HoldBound, ObjectReceipt, ReceiptBuilder, ReceiptChains, ReceiptVerifier, ReceiptsTrack, VerifiedGroup};
use joule_moq::{GroupId, ObjectId, ObjectStatus, TrackAlias, TrackName, TrackNamespace, JWP_MOQ_EXTENSION_VERSION};
use serde_json::{json, Value};
use std::collections::{BTreeMap, HashMap};
use wasm_bindgen::prelude::*;
/// The ALPN and `WT-Available-Protocols` value of draft-22 (§6.2).
pub const PROTOCOL: &str = "moqt-22";
fn js(e: impl std::fmt::Display) -> JsError {
JsError::new(&e.to_string())
}
use joule_moq::draft::Draft;
use joule_moq::draft23 as m23;
thread_local! {
/// The draft this page's session speaks on the wire; -22 until the page says otherwise.
static DRAFT: std::cell::Cell<Draft> = const { std::cell::Cell::new(Draft::D22) };
}
/// The draft encoders and decoders made from now on speak.
pub fn current_draft() -> Draft {
DRAFT.with(|d| d.get())
}
/// Set the draft the session speaks: 22 or 23, the drafts a page's transports carry.
pub fn set_draft_number(n: u32) -> Result<(), String> {
match Draft::from_number(n) {
Some(d @ (Draft::D22 | Draft::D23)) => {
DRAFT.with(|c| c.set(d));
Ok(())
}
_ => Err(format!("a page speaks drafts 23 and 22, not {n}")),
}
}
/// [`set_draft_number`] for a page, once its transport has negotiated the draft.
#[wasm_bindgen(js_name = setDraft)]
pub fn set_draft(n: u32) -> Result<(), JsError> {
set_draft_number(n).map_err(js)
}
/// The ALPN and `WT-Available-Protocols` value of the draft the page speaks (§6.2).
#[wasm_bindgen]
pub fn protocol() -> String {
current_draft().alpn_str().into()
}
/// A -22 message as the bytes of the draft the page speaks.
fn wire(m: &ControlMessage) -> Result<Vec<u8>, String> {
match current_draft() {
Draft::D23 => {
let mapped = m23::bridge::control_from22(m, m23::bridge::At::ControlStream).map_err(|u| u.to_string())?;
let m = mapped.msg.ok_or_else(|| format!("{} has no -23 form", m.name()))?;
m.encode().map_err(|e| e.to_string())
}
_ => m.encode().map_err(|e| e.to_string()),
}
}
// ---------------------------------------------------------------------------------------
// Control messages
// ---------------------------------------------------------------------------------------
/// A namespace from a JSON array of its fields, each a UTF-8 string.
fn namespace_of(v: &Value) -> Result<TrackNamespace, String> {
let fields = v.as_array().ok_or("a namespace is an array of strings")?;
let mut out = Vec::with_capacity(fields.len());
for f in fields {
out.push(f.as_str().ok_or("a namespace field is a string")?.to_owned());
}
Ok(TrackNamespace::from_strs(out.iter().map(String::as_str)))
}
fn u64_of(v: &Value, k: &str) -> Result<u64, String> {
v.get(k).and_then(Value::as_u64).ok_or_else(|| format!("`{k}` is an unsigned integer"))
}
fn location_of(v: &Value) -> Result<Location, String> {
let a = v.as_array().filter(|a| a.len() == 2).ok_or("a location is [group, object]")?;
match (a[0].as_u64(), a[1].as_u64()) {
(Some(g), Some(o)) => Ok(Location::new(g, o)),
_ => Err("a location is [group, object]".into()),
}
}
fn filter_of(v: &Value) -> Result<LocationFilter, String> {
let ty = v.get("type").and_then(Value::as_str).ok_or("a location filter names its `type`")?;
Ok(match ty {
"none" => LocationFilter::None,
"next" => LocationFilter::NextObject,
"relative" => LocationFilter::RelativeStart { start_group: u64_of(v, "start_group")? },
"absolute" => LocationFilter::AbsoluteStart { start: location_of(&v["start"])? },
"absolute_group_end" => LocationFilter::AbsoluteStartGroupEnd { start: location_of(&v["start"])?, end_group_delta: u64_of(v, "end_group_delta")? },
"range" => LocationFilter::AbsoluteRange { start: location_of(&v["start"])?, end_group_delta: u64_of(v, "end_group_delta")?, end_object: u64_of(v, "end_object")? },
other => return Err(format!("no location filter `{other}`")),
})
}
fn token_of(hex_value: &str) -> Result<Token, String> {
Ok(Token::UseValue { token_type: 0, value: hex::decode(hex_value).map_err(|e| format!("a token's value: {e}"))? })
}
/// Message Parameters from a JSON object (§9.20): `subscriber_priority`, `group_order`
/// (1 ascending, 2 descending), `forward`, `location_filter`, `object_delivery_timeout`,
/// `subgroup_delivery_timeout`, `rendezvous_timeout`, `fill_timeout`, `new_group_request`,
/// `authorization_token` (the hexadecimal value of a token of type 0, by value), and
/// `fill` (FILL_PARAMETERS, the same form).
fn params_of(v: &Value) -> Result<Parameters, String> {
let mut p = Parameters::new();
let Some(o) = v.as_object() else {
return if v.is_null() { Ok(p) } else { Err("parameters are an object".into()) };
};
for (k, x) in o {
let n = || x.as_u64().ok_or_else(|| format!("`{k}` is an unsigned integer"));
p = p.with(match k.as_str() {
"subscriber_priority" => Parameter::SubscriberPriority(u8::try_from(n()?).map_err(|_| "a priority is 0 to 255")?),
"group_order" => Parameter::GroupOrder(GroupOrder::from_u8(u8::try_from(n()?).unwrap_or(0), "group order").map_err(|e| e.to_string())?),
"forward" => Parameter::Forward(x.as_bool().ok_or("`forward` is a boolean")?),
"location_filter" => Parameter::LocationFilter(filter_of(x)?),
"object_delivery_timeout" => Parameter::ObjectDeliveryTimeout(n()?),
"subgroup_delivery_timeout" => Parameter::SubgroupDeliveryTimeout(n()?),
"rendezvous_timeout" => Parameter::RendezvousTimeout(n()?),
"fill_timeout" => Parameter::FillTimeout(n()?),
"new_group_request" => Parameter::NewGroupRequest(n()?),
"authorization_token" => Parameter::AuthorizationToken(token_of(x.as_str().ok_or("a token is hexadecimal")?)?),
"fill" => Parameter::FillParameters(params_of(x)?),
other => return Err(format!("no parameter `{other}` here")),
});
}
Ok(p)
}
/// Track Properties from a JSON object: `publisher_priority` (DEFAULT_PUBLISHER_PRIORITY)
/// and `object_delivery_timeout` (OBJECT_DELIVERY_TIMEOUT), in milliseconds.
fn track_props_of(v: &Value) -> Result<Properties, String> {
let mut p = Properties::new();
if let Some(x) = v.get("publisher_priority") {
p.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, x.as_u64().ok_or("`publisher_priority` is an integer")?));
}
if let Some(x) = v.get("object_delivery_timeout") {
p.push(KeyValuePair::varint(prop::OBJECT_DELIVERY_TIMEOUT, x.as_u64().ok_or("`object_delivery_timeout` is an integer")?));
}
// Rating claims, each a `WAI_RATING` Track Property (WAI age-assurance §5.7).
if let Some(x) = v.get("ratings") {
for c in x.as_array().ok_or("`ratings` is a list of claims")? {
p.push(joule_moq::rating::property(c.as_str().ok_or("a rating claim is its text")?));
}
}
Ok(p)
}
/// A rating claim (WAI age-assurance §3.1) by the principal whose seed is `seed` about its own
/// track `name` under the namespace `fields` (live-rooms §6.1), from group `from_group` on,
/// rated `min_age` under `scheme` with `value`, issued at the date-time `issued_at`: its text,
/// which a relay honours as the track owner's (§3.6).
#[allow(clippy::too_many_arguments)]
pub fn rating_claim_text(seed: [u8; 32], fields: &[&str], name: &str, from_group: u64, scheme: &str, value: &str, min_age: u64, issued_at: &str) -> Result<String, String> {
let ns = TrackNamespace::from_strs(fields.iter().copied());
let track = joule_moq::rating::track_digest(&ns, &TrackName::from_str(name));
let text = joule_moq::rating::write(&SigningKey::from_bytes(&seed), &joule_moq::rating::Subject::Track { track, from_group }, &[joule_moq::rating::Rating { scheme: scheme.into(), value: value.into(), min_age }], issued_at);
joule_moq::rating::read(text.as_bytes()).map_err(|r| format!("the claim does not read: {}", r.code()))?;
Ok(text)
}
/// [`rating_claim_text`] for a page: `fields` a JSON list of the namespace's fields.
#[wasm_bindgen(js_name = ratingClaim)]
#[allow(clippy::too_many_arguments)]
pub fn rating_claim(seed: &[u8], fields: &str, name: &str, from_group: u64, scheme: &str, value: &str, min_age: u64, issued_at: &str) -> Result<String, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
let fields: Vec<String> = serde_json::from_str(fields).map_err(js)?;
let f: Vec<&str> = fields.iter().map(String::as_str).collect();
rating_claim_text(seed, &f, name, from_group, scheme, value, min_age, issued_at).map_err(js)
}
/// The rating claims a property list carries (age-assurance §5.7): from the `track_properties`
/// of a SUBSCRIBE_OK in this crate's JSON view, the text of each `WAI_RATING`.
#[wasm_bindgen(js_name = ratingsOf)]
pub fn ratings_of(properties_json: &str) -> Result<String, JsError> {
let v: Value = serde_json::from_str(properties_json).map_err(js)?;
let out: Vec<String> = v
.as_array()
.into_iter()
.flatten()
.filter(|p| p["type"].as_u64() == Some(joule_moq::rating::WAI_RATING))
.filter_map(|p| p["bytes"].as_str().and_then(|h| hex::decode(h).ok()).and_then(|b| String::from_utf8(b).ok()))
.collect();
Ok(serde_json::to_string(&out).expect("JSON"))
}
/// The client's SETUP (§9.1): its implementation's name, MAX_REQUEST_UPDATES when not
/// zero, an authorization token of type 0 by value when given, and `joule-moq`'s receipt
/// extension when `receipts` (Setup Option `0x3A46`). Over WebTransport a client sends no
/// PATH or AUTHORITY, and none is sent here.
pub fn setup_bytes(token: Option<&[u8]>, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
let mut o = SetupOptions::new();
o.push(KeyValuePair::bytes(opt::MOQT_IMPLEMENTATION, implementation.as_bytes().to_vec()));
if max_request_updates != 0 {
o.push(KeyValuePair::varint(opt::MAX_REQUEST_UPDATES, max_request_updates));
}
if let Some(t) = token {
o.push(KeyValuePair::bytes(opt::AUTHORIZATION_TOKEN, Token::UseValue { token_type: 0, value: t.to_vec() }.to_bytes()));
}
if receipts {
o.push(KeyValuePair::varint(SETUP_OPTION_JWP_RECEIPTS, JWP_MOQ_EXTENSION_VERSION as u64));
}
wire(&ControlMessage::Setup(Setup { options: o }))
}
/// The bytes a client's control stream begins with: the stream type and its SETUP (§6.3),
/// which [`setup_bytes`] makes; SETUP's type is the control stream's type.
#[wasm_bindgen(js_name = clientSetup)]
pub fn client_setup(token: Option<Vec<u8>>, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
setup_bytes(token.as_deref(), implementation, receipts, max_request_updates).map_err(js)
}
/// The client's SETUP presenting a room grant (WAI live-rooms §10.5.2): the grant's record
/// as an AUTHORIZATION TOKEN (USE_VALUE, type 0) and its proof (`GRANT_PROOF`), signed by
/// the principal's key, from `seed`, over the relay's `challenge` for `target`, the
/// `//authority/path` the client addressed. A grant is public, so the proof is what makes
/// presenting it mean anything; the relay refuses a session that presents one without it.
pub fn setup_with_grant_bytes(record: &[u8], seed: [u8; 32], challenge: [u8; 32], target: &str, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
let (digest, principal) = grant_proof::grant_digest_and_principal(record).ok_or("not a WRG1 grant record")?;
let key = SigningKey::from_bytes(&seed);
if key.verifying_key().to_bytes() != principal {
return Err("the seed is not the grant's principal's".into());
}
let proof = GrantProof::sign(digest, &key, challenge, &format!("{} {target}", grant_proof::VERB_SETUP)).ok_or("a target a proof cannot carry")?;
setup_presenting(record, &proof.to_bytes(), implementation, receipts, max_request_updates)
}
/// The client's SETUP presenting `record` beside the proof record `proof`, as given.
pub fn setup_presenting(record: &[u8], proof: &[u8], implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, String> {
let proof = GrantProof::read(proof).map_err(|r| r.code().to_string())?;
let mut o = SetupOptions::new();
o.push(KeyValuePair::bytes(opt::MOQT_IMPLEMENTATION, implementation.as_bytes().to_vec()));
if max_request_updates != 0 {
o.push(KeyValuePair::varint(opt::MAX_REQUEST_UPDATES, max_request_updates));
}
grant_proof::add_grant(&mut o, record, &proof);
if receipts {
o.push(KeyValuePair::varint(SETUP_OPTION_JWP_RECEIPTS, JWP_MOQ_EXTENSION_VERSION as u64));
}
wire(&ControlMessage::Setup(Setup { options: o }))
}
/// [`setup_with_grant_bytes`] for a page: the challenge is a `GRANT_CHALLENGE` option
/// (`0x3A47`) read from a SETUP of the relay's on an earlier session, since the relay reads
/// this SETUP before it sends its own.
#[wasm_bindgen(js_name = clientSetupWithGrant)]
pub fn client_setup_with_grant(record: &[u8], seed: &[u8], challenge: &[u8], target: &str, implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
setup_with_grant_bytes(record, seed, challenge, target, implementation, receipts, max_request_updates).map_err(js)
}
/// A proof record (WAI live-rooms §10.5.1) for the grant `record`, by the key of `seed`
/// whatever key that is, over `challenge`, for `target`. For a test that a relay refuses a
/// proof by a key that is not the grant's principal (§10.5.4 step 4): a page proving its
/// own grant uses [`client_setup_with_grant`], which signs with the principal's key alone.
#[wasm_bindgen(js_name = grantProofBy)]
pub fn grant_proof_by(record: &[u8], seed: &[u8], challenge: &[u8], target: &str) -> Result<Vec<u8>, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
let (digest, _) = grant_proof::grant_digest_and_principal(record).ok_or_else(|| JsError::new("not a WRG1 grant record"))?;
let p = GrantProof::sign(digest, &SigningKey::from_bytes(&seed), challenge, &format!("{} {target}", grant_proof::VERB_SETUP)).ok_or_else(|| JsError::new("a target a proof cannot carry"))?;
Ok(p.to_bytes())
}
/// A proof record (WAI live-rooms §10.5.1) for the grant `record` by its principal, the key of
/// `seed`, over `challenge`, for an HTTP request of method `verb` to `target` (`//authority`
/// and the path the page addressed): what the request's `WAI-Room-Proof` carries (§10.5.3). A
/// seed whose key is not the grant's principal makes no proof.
pub fn http_proof_bytes(record: &[u8], seed: [u8; 32], challenge: [u8; 32], verb: &str, target: &str) -> Result<Vec<u8>, String> {
let (digest, principal) = grant_proof::grant_digest_and_principal(record).ok_or("not a WRG1 grant record")?;
let key = SigningKey::from_bytes(&seed);
if key.verifying_key().to_bytes() != principal {
return Err("the seed is not the grant's principal's".into());
}
let p = GrantProof::sign(digest, &key, challenge, &format!("{verb} {target}")).ok_or("a target a proof cannot carry")?;
Ok(p.to_bytes())
}
/// [`http_proof_bytes`] for a page.
#[wasm_bindgen(js_name = grantProofHttp)]
pub fn grant_proof_http(record: &[u8], seed: &[u8], challenge: &[u8], verb: &str, target: &str) -> Result<Vec<u8>, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
http_proof_bytes(record, seed, challenge, verb, target).map_err(js)
}
/// A proof record (WAI live-rooms §10.5.1) naming no grant (its `grant` 32 zero bytes), by the
/// key of `seed`, over `challenge`, for an HTTP request of method `verb` to `target`: what a
/// request proving a key alone carries in `WAI-Room-Proof`, as a conversation's join and a
/// call's inbox take it (WAI live-messaging §10.4).
#[wasm_bindgen(js_name = keyProofHttp)]
pub fn key_proof_http(seed: &[u8], challenge: &[u8], verb: &str, target: &str) -> Result<Vec<u8>, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
let challenge: [u8; 32] = challenge.try_into().map_err(|_| JsError::new("a challenge is 32 bytes"))?;
let p = GrantProof::sign([0; 32], &SigningKey::from_bytes(&seed), challenge, &format!("{verb} {target}")).ok_or_else(|| JsError::new("a target a proof cannot carry"))?;
Ok(p.to_bytes())
}
/// [`setup_presenting`] for a page.
#[wasm_bindgen(js_name = clientSetupWithProof)]
pub fn client_setup_with_proof(record: &[u8], proof: &[u8], implementation: &str, receipts: bool, max_request_updates: u64) -> Result<Vec<u8>, JsError> {
setup_presenting(record, proof, implementation, receipts, max_request_updates).map_err(js)
}
/// The Setup Option type of the relay's challenge (WAI live-rooms §10.5.2).
#[wasm_bindgen(js_name = grantChallengeOption)]
pub fn grant_challenge_option() -> u64 {
grant_proof::SETUP_OPTION_GRANT_CHALLENGE
}
/// The Setup Option type by which a relay says it refuses the session at SETUP, its value
/// the reason's code (WAI live-rooms §10.5.2).
#[wasm_bindgen(js_name = grantRefusedOption)]
pub fn grant_refused_option() -> u64 {
grant_proof::SETUP_OPTION_GRANT_REFUSED
}
/// A control message a client sends, from its JSON description: `message` is its name in
/// draft-22's Table 5, and the rest are its fields:
///
/// - `SUBSCRIBE`, `FETCH`, `TRACK_STATUS`: `request_id`, `namespace` (an array of UTF-8
/// fields), `name` (UTF-8) and `params` (see [`params_of`]);
/// - `PUBLISH`: those, `track_alias` and `track_properties`;
/// - `PUBLISH_NAMESPACE`, `SUBSCRIBE_NAMESPACE`: `request_id`, `prefix` and `params`;
/// - `REQUEST_OK`: `params` and `track_properties`; `SUBSCRIBE_OK`: `track_alias` too;
/// - `REQUEST_UPDATE`: `request_id`, `params`;
/// - `REQUEST_ERROR`: `code`, `retry_interval`, `reason`;
/// - `PUBLISH_DONE`: `status`, `stream_count`, `reason`.
pub fn control_bytes(desc: &str) -> Result<Vec<u8>, String> {
let v: Value = serde_json::from_str(desc).map_err(|e| format!("a message description: {e}"))?;
let name = v.get("message").and_then(Value::as_str).ok_or("a message description names its `message`")?;
let params = || params_of(v.get("params").unwrap_or(&Value::Null));
let props = || track_props_of(v.get("track_properties").unwrap_or(&Value::Null));
let track = || -> Result<TrackRequest, String> {
Ok(TrackRequest {
request_id: u64_of(&v, "request_id")?,
namespace: namespace_of(&v["namespace"])?,
name: TrackName::from_str(v.get("name").and_then(Value::as_str).ok_or("`name` is a string")?),
params: params()?,
})
};
let ns_req = || -> Result<NamespaceRequest, String> { Ok(NamespaceRequest { request_id: u64_of(&v, "request_id")?, prefix: namespace_of(&v["prefix"])?, params: params()? }) };
let reason = || v.get("reason").and_then(Value::as_str).unwrap_or("").as_bytes().to_vec();
let m = match name {
"SUBSCRIBE" => ControlMessage::Subscribe(track()?),
"FETCH" => ControlMessage::Fetch(track()?),
"TRACK_STATUS" => ControlMessage::TrackStatus(track()?),
"PUBLISH" => {
let t = track()?;
ControlMessage::Publish(Publish { request_id: t.request_id, namespace: t.namespace, name: t.name, track_alias: u64_of(&v, "track_alias")?, params: t.params, track_properties: props()? })
}
"PUBLISH_NAMESPACE" => ControlMessage::PublishNamespace(ns_req()?),
"SUBSCRIBE_NAMESPACE" => ControlMessage::SubscribeNamespace(ns_req()?),
"REQUEST_OK" => ControlMessage::RequestOk(RequestOk { params: params()?, track_properties: props()? }),
"SUBSCRIBE_OK" => ControlMessage::SubscribeOk(SubscribeOk { track_alias: u64_of(&v, "track_alias")?, params: params()?, track_properties: props()? }),
"REQUEST_UPDATE" => ControlMessage::RequestUpdate(RequestUpdate { request_id: u64_of(&v, "request_id")?, params: params()? }),
"REQUEST_ERROR" => ControlMessage::RequestError(RequestError {
code: RequestErrorCode(u64_of(&v, "code")?),
retry_interval: v.get("retry_interval").and_then(Value::as_u64).unwrap_or(0),
reason: reason(),
redirect: None,
}),
"PUBLISH_DONE" => ControlMessage::PublishDone(PublishDone {
status: PublishDoneCode(u64_of(&v, "status")?),
stream_count: v.get("stream_count").and_then(Value::as_u64).unwrap_or(u64::MAX),
reason: reason(),
}),
other => return Err(format!("a client does not send `{other}` here")),
};
wire(&m)
}
#[wasm_bindgen(js_name = encodeControl)]
pub fn encode_control(desc: &str) -> Result<Vec<u8>, JsError> {
control_bytes(desc).map_err(js)
}
/// A refusal as JSON: its kind and the consequence draft-22 gives it.
fn wire_error(e: &WireError) -> String {
json::error(e).to_string()
}
/// A refusal as JSON with the consequence its draft gives it: at -23, the code -23 gives the
/// fault on `plane` (`joule_moq::draft23::error`).
fn wire_error_at(e: &WireError, draft: Draft, plane: m23::Plane) -> String {
match draft {
Draft::D23 => m23::json::error(e, plane).to_string(),
_ => wire_error(e),
}
}
/// Framed control messages (§9: Type, a Length, the body: 16 bits at -22, a varint at -23)
/// read from a request stream as its bytes arrive. Each [`MessageStream::next`] gives the
/// next whole message as JSON (`draft22::json::control`, whichever draft is on the wire),
/// or nothing until more bytes come.
#[wasm_bindgen]
pub struct MessageStream {
buf: Vec<u8>,
draft: Draft,
/// The request this stream carries, when the page said ([`MessageStream::for_request`]):
/// at -23 a FETCH's answer is a REQUEST_OK, which the page reads as FETCH_OK.
kind: Option<joule_moq::draft22::RequestKind>,
fetch: Option<Option<LocationFilter>>,
answered: bool,
}
impl Default for MessageStream {
fn default() -> Self {
MessageStream { buf: Vec::new(), draft: current_draft(), kind: None, fetch: None, answered: false }
}
}
impl MessageStream {
fn framed_len(b: &[u8]) -> Option<usize> {
let first = *b.first()?;
let tlen = joule_moq::draft22::varint::len_from_first(first);
if b.len() < tlen + 2 {
return None;
}
Some(tlen + 2 + u16::from_be_bytes([b[tlen], b[tlen + 1]]) as usize)
}
fn framed_len23(b: &[u8]) -> Option<usize> {
let (_, tl) = joule_moq::draft22::varint::get(b, "").ok()?;
let (len, ll) = joule_moq::draft22::varint::get(&b[tl..], "").ok()?;
tl.checked_add(ll)?.checked_add(usize::try_from(len).ok()?)
}
pub fn next_message(&mut self) -> Result<Option<ControlMessage>, WireError> {
if self.draft == Draft::D23 {
return self.next_message23();
}
match Self::framed_len(&self.buf).filter(|n| *n <= self.buf.len()) {
Some(n) => {
let m = ControlMessage::decode_exact(&self.buf[..n])?;
self.buf.drain(..n);
Ok(Some(m))
}
None => Ok(None),
}
}
fn next_message23(&mut self) -> Result<Option<ControlMessage>, WireError> {
// A Length past MAX_CONTROL_MESSAGE_SIZE is refused before its body is waited for.
if let Ok((_, tl)) = joule_moq::draft22::varint::get(&self.buf, "")
&& let Ok((len, _)) = joule_moq::draft22::varint::get(&self.buf[tl..], "")
&& len > m23::control::DEFAULT_MAX_CONTROL_MESSAGE_SIZE
{
return Err(WireError::LengthExceedsLimit { what: "control message", len, limit: m23::control::DEFAULT_MAX_CONTROL_MESSAGE_SIZE });
}
let Some(n) = Self::framed_len23(&self.buf).filter(|n| *n <= self.buf.len()) else { return Ok(None) };
let m = m23::ControlMessage::decode_exact(&self.buf[..n])?;
self.buf.drain(..n);
let at = match self.kind {
Some(kind) => m23::bridge::At::Request { kind, answers_update: self.answered && matches!(m, m23::ControlMessage::RequestOk(_) | m23::ControlMessage::RequestError(_)), fetch: self.fetch },
None => m23::bridge::At::ControlStream,
};
if matches!(m, m23::ControlMessage::RequestOk(_) | m23::ControlMessage::RequestError(_) | m23::ControlMessage::SubscribeOk(_)) {
self.answered = true;
}
let mapped = m23::bridge::control_to22(&m, at).map_err(|u| WireError::Unencodable { what: "a -23 message", reason: u.reason })?;
Ok(Some(mapped.msg))
}
}
#[wasm_bindgen]
impl MessageStream {
#[wasm_bindgen(constructor)]
pub fn new() -> MessageStream {
MessageStream::default()
}
/// The request this stream answers, as the description [`encode_control`] took: at -23
/// the answer to a FETCH is read as FETCH_OK, its End Location from LARGEST_OBJECT or the
/// range the FETCH asked for.
#[wasm_bindgen(js_name = forRequest)]
pub fn for_request(&mut self, desc: &str) -> Result<(), JsError> {
use joule_moq::draft22::RequestKind as K;
let v: Value = serde_json::from_str(desc).map_err(js)?;
self.kind = match v.get("message").and_then(Value::as_str) {
Some("SUBSCRIBE") => Some(K::Subscribe),
Some("FETCH") => Some(K::Fetch),
Some("TRACK_STATUS") => Some(K::TrackStatus),
Some("PUBLISH") => Some(K::Publish),
Some("PUBLISH_NAMESPACE") => Some(K::PublishNamespace),
Some("SUBSCRIBE_NAMESPACE") => Some(K::SubscribeNamespace),
_ => None,
};
if self.kind == Some(K::Fetch) {
self.fetch = Some(params_of(v.get("params").unwrap_or(&Value::Null)).map_err(js)?.location_filter());
}
Ok(())
}
pub fn push(&mut self, bytes: &[u8]) {
self.buf.extend_from_slice(bytes);
}
/// The next whole message as JSON, or `undefined`. (`next` is the page's name for
/// it; this is not an iterator.)
#[allow(clippy::should_implement_trait)]
pub fn next(&mut self) -> Result<Option<String>, JsError> {
let draft = self.draft;
self.next_message().map(|m| m.map(|m| json::control(&m).to_string())).map_err(|e| JsError::new(&wire_error_at(&e, draft, m23::Plane::Control)))
}
/// Bytes held that are not yet a whole message: a stream that ends with any is cut
/// inside a message.
pub fn buffered(&self) -> usize {
self.buf.len()
}
}
// ---------------------------------------------------------------------------------------
// Data streams
// ---------------------------------------------------------------------------------------
/// One item read from a unidirectional stream: `kind` is `header`, `object`,
/// `fetch_object`, `end_of_range`, `control` or `padding`; `json` describes it without its
/// payload; [`UniItem::payload`] is the object's payload.
#[wasm_bindgen]
pub struct UniItem {
kind: String,
json: String,
payload: Vec<u8>,
properties: Option<Properties>,
}
#[wasm_bindgen]
impl UniItem {
#[wasm_bindgen(getter)]
pub fn kind(&self) -> String {
self.kind.clone()
}
#[wasm_bindgen(getter)]
pub fn json(&self) -> String {
self.json.clone()
}
/// The payload, moved out: a second call gives an empty array.
pub fn payload(&mut self) -> Vec<u8> {
std::mem::take(&mut self.payload)
}
/// An object's property list as the stream carried it, encoded (MOQT §8.3): what a
/// member opens a sealed object with, its Immutable Properties byte for byte. Empty for
/// an object with none, and for an item that is not an object.
pub fn properties(&self) -> Vec<u8> {
self.properties.as_ref().and_then(|p| p.to_bytes().ok()).unwrap_or_default()
}
}
fn item_of(e: UniEvent) -> UniItem {
match e {
UniEvent::SubgroupObject(o) => UniItem {
kind: "object".into(),
json: json!({
"object_id": o.object_id,
"status": o.status.code(),
"properties": json::properties(&o.properties),
})
.to_string(),
payload: o.payload,
properties: Some(o.properties),
},
UniEvent::FetchEntry(FetchEntry::Object(o)) => UniItem {
kind: "fetch_object".into(),
json: json!({
"group_id": o.group_id,
"object_id": o.object_id,
"subgroup_id": o.subgroup_id,
"publisher_priority": o.publisher_priority,
})
.to_string(),
payload: o.payload,
properties: o.properties,
},
other => {
let v = json::uni_event(&other);
let kind = match &other {
UniEvent::Header(_) => "header",
UniEvent::Control(_) => "control",
UniEvent::FetchEntry(_) => "end_of_range",
UniEvent::Padding(_) => "padding",
_ => "other",
};
UniItem { kind: kind.into(), json: v.to_string(), payload: Vec::new(), properties: None }
}
}
}
/// The items of one unidirectional stream (a control, subgroup, fetch or padding stream),
/// each given once all of it has arrived: `joule-moq`'s incremental decoder, bounded, of
/// the draft the page speaks; a -23 stream's items are given in -22's model.
#[wasm_bindgen]
pub struct UniDecoder {
dec: UniInner,
}
enum UniInner {
D22(UniStreamDecoder),
D23(Box<m23::UniStreamDecoder>),
}
fn header23(h: m23::StreamHeader) -> StreamHeader {
match h {
m23::StreamHeader::Control => StreamHeader::Control,
m23::StreamHeader::Subgroup(s) => StreamHeader::Subgroup(s),
m23::StreamHeader::Fetch(f) => StreamHeader::Fetch(f),
m23::StreamHeader::Padding => StreamHeader::Padding,
}
}
impl UniDecoder {
fn draft(&self) -> Draft {
match self.dec {
UniInner::D22(_) => Draft::D22,
UniInner::D23(_) => Draft::D23,
}
}
/// The next item, in -22's model.
pub fn next_event(&mut self) -> Result<Option<UniEvent>, WireError> {
match &mut self.dec {
UniInner::D22(d) => d.next(),
UniInner::D23(d) => loop {
let Some(e) = d.next()? else { return Ok(None) };
return Ok(Some(match e {
m23::UniEvent::Header(h) => UniEvent::Header(header23(h)),
m23::UniEvent::Control(m) => UniEvent::Control(
m23::bridge::control_to22(&m, m23::bridge::At::ControlStream).map_err(|u| WireError::Unencodable { what: "a -23 message", reason: u.reason })?.msg,
),
m23::UniEvent::SubgroupObject(o) => UniEvent::SubgroupObject(o),
m23::UniEvent::FetchEntry(f) => match m23::bridge::fetch_entry_to22(&f).msg {
Some(f) => UniEvent::FetchEntry(f),
// End of Group and End of Track: -22 has no entry for them.
None => continue,
},
m23::UniEvent::Padding(n) => UniEvent::Padding(n),
}));
},
}
}
}
#[wasm_bindgen]
impl UniDecoder {
/// A decoder that refuses an object payload longer than `max_payload` bytes.
#[wasm_bindgen(constructor)]
pub fn new(max_payload: u64) -> UniDecoder {
let limits = Limits { max_payload, ..Limits::default() };
let dec = match current_draft() {
Draft::D23 => UniInner::D23(Box::new(m23::UniStreamDecoder::new(GroupOrder::Ascending, limits))),
_ => UniInner::D22(UniStreamDecoder::new(GroupOrder::Ascending, limits)),
};
UniDecoder { dec }
}
/// The Group Order of the FETCH a fetch stream answers: 1 ascending, 2 descending.
#[wasm_bindgen(js_name = setGroupOrder)]
pub fn set_group_order(&mut self, order: u8) {
let o = if order == 2 { GroupOrder::Descending } else { GroupOrder::Ascending };
match &mut self.dec {
UniInner::D22(d) => d.set_group_order(o),
UniInner::D23(d) => d.set_group_order(o),
}
}
pub fn push(&mut self, bytes: &[u8]) {
match &mut self.dec {
UniInner::D22(d) => d.push(bytes),
UniInner::D23(d) => d.push(bytes),
}
}
#[allow(clippy::should_implement_trait)]
pub fn next(&mut self) -> Result<Option<UniItem>, JsError> {
let draft = self.draft();
self.next_event().map(|e| e.map(item_of)).map_err(|e| JsError::new(&wire_error_at(&e, draft, m23::Plane::Stream)))
}
/// At the stream's FIN: a stream that ends inside an item is refused.
pub fn finish(&self) -> Result<(), JsError> {
match &self.dec {
UniInner::D22(d) => d.finish(),
UniInner::D23(d) => d.finish(),
}
.map_err(|e| JsError::new(&wire_error_at(&e, self.draft(), m23::Plane::Stream)))
}
}
/// A subgroup stream (§11.4), written as its objects come: [`SubgroupWriter::header`]
/// once, then [`SubgroupWriter::object`] for each object, the Object IDs increasing.
#[wasm_bindgen]
pub struct SubgroupWriter {
header: SubgroupHeader,
codec: SubgroupCodec,
}
#[wasm_bindgen]
impl SubgroupWriter {
/// Subgroup 0 of group `group_id` of the track whose alias is `track_alias`, at
/// `priority`; `end_of_group` when the stream's FIN ends the group.
#[wasm_bindgen(constructor)]
pub fn new(track_alias: u64, group_id: u64, priority: u8, end_of_group: bool) -> SubgroupWriter {
let header = SubgroupHeader { track_alias, group_id, subgroup_id: SubgroupId::Zero, publisher_priority: Some(priority), properties: false, end_of_group, first_object: true };
SubgroupWriter { header, codec: SubgroupCodec::new(header) }
}
pub fn header(&self) -> Vec<u8> {
let mut b = Vec::new();
self.header.encode(&mut b);
b
}
pub fn object(&mut self, object_id: u64, payload: &[u8]) -> Result<Vec<u8>, JsError> {
let mut b = Vec::new();
self.codec.encode_object(&SubgroupObject::normal(object_id, payload.to_vec()), &mut b).map_err(|e| JsError::new(&wire_error(&e)))?;
Ok(b)
}
}
// ---------------------------------------------------------------------------------------
// Receipts: the publisher
// ---------------------------------------------------------------------------------------
/// The unmetered marker's signed class (WAI energy-measurement §2.7, tag 0): a page has no
/// meter, and its figure is never a measurement of zero joules.
const UNMETERED: [u8; 1] = [0];
/// One media group signed and written: the receipts-track stream and the media stream.
pub struct Signed {
pub receipts_stream: Vec<u8>,
pub media_stream: Vec<u8>,
pub content_hash: [u8; 32],
pub link: [u8; 32],
}
/// The publisher's side of one media track and its receipts track (WAI
/// moq-streaming-format §4.1): each group is one object, signed into its own group
/// receipt, whose parent is the group before it, so the track's receipts form one chain.
pub struct Signer {
key: SigningKey,
receipt_alias: u64,
media_alias: u64,
receipts_alias: u64,
priority: u8,
origin: String,
object_timeout_ms: Option<u64>,
parent: Option<[u8; 32]>,
/// Whether each object is one WAI envelope, flagged so (`WAI_ENVELOPE`): a media track's
/// are; a world track's op batches are not (WAI live-worlds §4.2).
envelope: bool,
}
impl Signer {
pub fn new(seed: [u8; 32], receipt_alias: u64, media_alias: u64, receipts_alias: u64, priority: u8, origin: String, object_timeout_ms: Option<u64>) -> Self {
Signer { key: SigningKey::from_bytes(&seed), receipt_alias, media_alias, receipts_alias, priority, origin, object_timeout_ms, parent: None, envelope: true }
}
/// Objects that are not WAI envelopes from here on: no `WAI_ENVELOPE` flag.
pub fn set_envelope(&mut self, envelope: bool) {
self.envelope = envelope;
}
pub fn public_key(&self) -> [u8; 32] {
self.key.verifying_key().to_bytes()
}
/// Group `group` of the media track: one object, `payload`, its receipt signed under
/// the publisher's key with the unmetered marker as its figure; the receipts-track
/// object of the same group; and the media object carrying its `WAI_RECEIPT` property
/// (moq-streaming-format §5) and the `WAI_ENVELOPE` flag.
pub fn sign(&mut self, group: u64, payload: &[u8], started: &str, ended: &str) -> Result<Signed, String> {
self.sign_with(group, payload, None, started, ended)
}
/// Group `group` of a sealed media track (WAI confidentiality §4, §8.1): `payload` is
/// the sealed object's payload, ciphertext and tag, and the receipt is over those bytes;
/// `properties` are the object's as its seal made them (the Immutable Properties with
/// its Key ID and envelope flag, the delivery timeout beside them). A sealed object
/// carries no `WAI_RECEIPT` (§4.3): its receipt rides the receipts track alone.
pub fn sign_sealed(&mut self, group: u64, payload: &[u8], properties: Properties, started: &str, ended: &str) -> Result<Signed, String> {
self.sign_with(group, payload, Some(properties), started, ended)
}
fn sign_with(&mut self, group: u64, payload: &[u8], sealed: Option<Properties>, started: &str, ended: &str) -> Result<Signed, String> {
let hash = ObjectReceipt::hash_payload(payload);
let mut b = ReceiptBuilder::new(TrackAlias(self.receipt_alias), GroupId(group));
b.add_labelled_object(hash, 0, self.origin.clone(), started, UNMETERED.to_vec()).map_err(|e| e.to_string())?;
let receipt = ObjectReceipt {
track_alias: TrackAlias(self.receipt_alias),
group_id: GroupId(group),
object_id: ObjectId(0),
content_hash: hash,
joules_micro: 0,
origin: self.origin.clone(),
merkle_proof: b.proof_for(0).map_err(|e| e.to_string())?,
energy_class: Some(UNMETERED.to_vec()),
};
let g = b.finalize_labelled(&self.key, self.origin.clone(), ended, self.parent).map_err(|e| e.to_string())?;
let link = g.link_hash();
let wai = WaiReceiptProperty::from_receipts(&receipt, &g).ok_or("the receipt gives no valid WAI_RECEIPT property")?;
let bundle = ReceiptBundle { group: g, objects: vec![receipt] };
let receipts_stream = bundle.encode_stream(self.receipts_alias, group, self.priority).map_err(|e| e.to_string())?;
let header = ObjectHeader {
track_alias: TrackAlias(self.media_alias),
group_id: GroupId(group),
object_id: ObjectId(0),
publisher_priority: self.priority,
status: ObjectStatus::Normal,
payload_len: payload.len() as u64,
mrl_layer: 0,
mrl_total_layers: 0,
wai_envelope: self.envelope,
};
let props = match sealed {
Some(p) => p,
None => object_properties(&header, &ObjectExtras { receipt: None, wai_receipt: Some(wai), delivery_timeout_ms: self.object_timeout_ms }).map_err(|e| e.to_string())?,
};
let sh = SubgroupHeader { track_alias: self.media_alias, group_id: group, subgroup_id: SubgroupId::Zero, publisher_priority: Some(self.priority), properties: true, end_of_group: true, first_object: true };
let mut media_stream = Vec::with_capacity(payload.len() + 256);
sh.encode(&mut media_stream);
SubgroupCodec::new(sh)
.encode_object(&SubgroupObject { object_id: 0, properties: props, status: WireStatus::Normal, payload: payload.to_vec() }, &mut media_stream)
.map_err(|e| e.to_string())?;
self.parent = Some(link);
Ok(Signed { receipts_stream, media_stream, content_hash: hash, link })
}
}
/// A [`Signed`] group handed to the page.
#[wasm_bindgen]
pub struct SignedGroup {
receipts_stream: Vec<u8>,
media_stream: Vec<u8>,
content_hash: String,
}
#[wasm_bindgen]
impl SignedGroup {
/// The receipts-track stream, moved out. It goes no later than the media stream.
#[wasm_bindgen(js_name = receiptsStream)]
pub fn receipts_stream(&mut self) -> Vec<u8> {
std::mem::take(&mut self.receipts_stream)
}
/// The media stream, moved out.
#[wasm_bindgen(js_name = mediaStream)]
pub fn media_stream(&mut self) -> Vec<u8> {
std::mem::take(&mut self.media_stream)
}
#[wasm_bindgen(getter, js_name = contentHash)]
pub fn content_hash(&self) -> String {
self.content_hash.clone()
}
}
/// The publisher's signer of one media track, for a page.
#[wasm_bindgen]
pub struct TrackSigner {
inner: Signer,
}
#[wasm_bindgen]
impl TrackSigner {
/// `seed`: the publisher's 32-byte Ed25519 seed. `receipt_alias`: the media track's
/// `wai_receipt_alias`. `media_alias` and `receipts_alias`: the Track Aliases this
/// session's PUBLISH gave the two tracks. `object_timeout_ms`: an
/// OBJECT_DELIVERY_TIMEOUT for each media object, 0 for none.
#[wasm_bindgen(constructor)]
pub fn new(seed: &[u8], receipt_alias: u64, media_alias: u64, receipts_alias: u64, priority: u8, origin: String, object_timeout_ms: u64) -> Result<TrackSigner, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
Ok(TrackSigner { inner: Signer::new(seed, receipt_alias, media_alias, receipts_alias, priority, origin, (object_timeout_ms > 0).then_some(object_timeout_ms)) })
}
#[wasm_bindgen(js_name = publicKey)]
pub fn public_key(&self) -> Vec<u8> {
self.inner.public_key().to_vec()
}
/// `false`: the track's objects are not WAI envelopes (a world track's op batches, WAI
/// live-worlds §4.2), and carry no `WAI_ENVELOPE` flag.
#[wasm_bindgen(js_name = setEnvelope)]
pub fn set_envelope(&mut self, envelope: bool) {
self.inner.set_envelope(envelope);
}
/// Re-alias after a new session (a GOAWAY): the chain continues.
#[wasm_bindgen(js_name = setAliases)]
pub fn set_aliases(&mut self, media_alias: u64, receipts_alias: u64) {
self.inner.media_alias = media_alias;
self.inner.receipts_alias = receipts_alias;
}
pub fn sign(&mut self, group: u64, payload: &[u8], started: &str, ended: &str) -> Result<SignedGroup, JsError> {
let s = self.inner.sign(group, payload, started, ended).map_err(js)?;
Ok(SignedGroup { receipts_stream: s.receipts_stream, media_stream: s.media_stream, content_hash: hex::encode(s.content_hash) })
}
/// [`Signer::sign_sealed`] for a page: `properties` as a sealed object's
/// [`e2ee::SealedOut::properties`] gives them.
#[cfg(feature = "e2ee")]
#[wasm_bindgen(js_name = signSealed)]
pub fn sign_sealed(&mut self, group: u64, payload: &[u8], properties: &[u8], started: &str, ended: &str) -> Result<SignedGroup, JsError> {
let props = Properties::decode(properties, "properties").map_err(|e| JsError::new(&wire_error(&e)))?;
let s = self.inner.sign_sealed(group, payload, props, started, ended).map_err(js)?;
Ok(SignedGroup { receipts_stream: s.receipts_stream, media_stream: s.media_stream, content_hash: hex::encode(s.content_hash) })
}
}
/// The public key of a 32-byte Ed25519 seed.
#[wasm_bindgen(js_name = publicKeyOf)]
pub fn public_key_of(seed: &[u8]) -> Result<Vec<u8>, JsError> {
let seed: [u8; 32] = seed.try_into().map_err(|_| JsError::new("a seed is 32 bytes"))?;
Ok(SigningKey::from_bytes(&seed).verifying_key().to_bytes().to_vec())
}
// ---------------------------------------------------------------------------------------
// Receipts: the subscriber
// ---------------------------------------------------------------------------------------
/// What the subscriber holds of one group of the media track once its receipts arrived
/// and verified: the verified root and the object receipts.
struct HeldGroup {
verified: VerifiedGroup,
objects: Vec<ObjectReceipt>,
/// The group receipt's link hash (WAI jwp-receipts §3 step 3): what a frame decoded from
/// one of its objects names as its `receipt` (WAI browser-sink §7).
link: [u8; 32],
}
/// The subscriber's side of one media track and its receipts track: each receipts-track
/// object is admitted to the track's chain (WAI jwp-receipts §3 step 3, as a subscriber
/// that joins: the chain is anchored at the first group it receives) after its signature
/// verifies under the publisher's key, and each media object is verified against the
/// receipt of its position.
pub struct Verifier {
track: ReceiptsTrack,
verifier: ReceiptVerifier,
chains: ReceiptChains,
groups: BTreeMap<u64, HeldGroup>,
admissions: HashMap<&'static str, u64>,
refusals: HashMap<String, u64>,
/// The size of each receipts-track object received, in the binary form, and of the
/// same receipts in the JSON form this crate's serialisation gives them (what the
/// receipts track carried before the binary form): a measurement, nothing more.
sizes: Sizes,
json_sizes: Sizes,
}
/// A count, the least, the most and the sum of some sizes in bytes.
#[derive(Default)]
struct Sizes {
n: u64,
min: u64,
max: u64,
sum: u64,
}
impl Sizes {
fn add(&mut self, b: u64) {
self.min = if self.n == 0 { b } else { self.min.min(b) };
self.max = self.max.max(b);
self.n += 1;
self.sum += b;
}
fn json(&self) -> Value {
json!({ "objects": self.n, "min": self.min, "max": self.max, "sum": self.sum })
}
}
/// The length of a bundle in this crate's JSON serialisation of its receipts, an array of
/// the group receipt and the object receipts.
fn json_form_len(b: &ReceiptBundle) -> u64 {
let mut items = vec![serde_json::to_value(&b.group).unwrap_or(Value::Null)];
items.extend(b.objects.iter().map(|o| serde_json::to_value(o).unwrap_or(Value::Null)));
serde_json::to_vec(&items).map_or(0, |v| v.len() as u64)
}
fn admission_name(a: Admission) -> &'static str {
match a {
Admission::Linked => "linked",
Admission::Anchor => "anchor",
Admission::History => "history",
Admission::AlreadyAdmitted => "already",
}
}
fn refusal_name(r: ChainRefusal) -> &'static str {
match r {
ChainRefusal::Unverified => "unverified",
ChainRefusal::MoqtMismatch => "moqt_mismatch",
ChainRefusal::ParentNotHeld => "parent_not_held",
ChainRefusal::ChainBreak => "chain_break",
ChainRefusal::GroupIdNotIncreasing => "group_id_not_increasing",
ChainRefusal::ParentNotRecovered => "parent_not_recovered",
}
}
/// The bound this subscriber states on what it holds for missing parents (jwp-receipts
/// §3 step 3, *Holding*).
pub const HOLD: HoldBound = HoldBound { max_groups: 512, max_history: 4096, max_total: 8192, max_age_ms: Some(30_000) };
impl Verifier {
pub fn new(publisher_key: [u8; 32], receipt_alias: u64) -> Result<Self, String> {
let key = ed25519_dalek::VerifyingKey::from_bytes(&publisher_key).map_err(|e| format!("the publisher's key: {e}"))?;
let chains = ReceiptChains::for_moq(HOLD).map_err(|b| format!("a hold bound of {}", b.count))?.with_join();
Ok(Verifier {
track: ReceiptsTrack { publisher_key, track_alias: TrackAlias(receipt_alias) },
verifier: ReceiptVerifier::with_anchors(vec![key]),
chains,
groups: BTreeMap::new(),
admissions: HashMap::new(),
refusals: HashMap::new(),
sizes: Sizes::default(),
json_sizes: Sizes::default(),
})
}
fn refuse(&mut self, why: &str) -> Value {
*self.refusals.entry(why.to_owned()).or_default() += 1;
json!({"ok": false, "reason": why})
}
/// Group `group` of the receipts track, its payload: the group receipt verified under
/// the publisher's key (and nothing else), every receipt's coordinates checked against
/// the MOQT group and Object ID it is for (`moqt_mismatch` otherwise), and the group
/// admitted to its chain. Groups held behind it that its admission released are
/// reported in `released`.
pub fn receipts(&mut self, group: u64, payload: &[u8], now_ms: u64) -> Value {
let bundle = match ReceiptBundle::from_payload(payload) {
Ok(b) => b,
Err(_) => return self.refuse("bundle_malformed"),
};
self.sizes.add(payload.len() as u64);
self.json_sizes.add(json_form_len(&bundle));
if bundle.objects.iter().enumerate().any(|(i, o)| o.check_moqt(self.track.track_alias, GroupId(group), ObjectId(i as u64)).is_err()) {
return self.refuse("moqt_mismatch");
}
let verified = match self.verifier.verify_group(&bundle.group) {
Ok(v) => v,
Err(_) => return self.refuse("unverified"),
};
if verified.group_id != GroupId(group) || verified.track_alias != self.track.track_alias {
return self.refuse("moqt_mismatch");
}
for o in &bundle.objects {
if self.verifier.verify_object(o, &verified).is_err() {
return self.refuse("object_receipt_invalid");
}
}
let chain = match self.chains.admit_on(&self.track, GroupId(group), &bundle.group, now_ms) {
Ok(a) => {
*self.admissions.entry(admission_name(a)).or_default() += 1;
json!(admission_name(a))
}
Err(r) => {
*self.refusals.entry(format!("chain:{}", refusal_name(r))).or_default() += 1;
json!(format!("held:{}", refusal_name(r)))
}
};
let released: Vec<Value> = self
.chains
.take_events()
.into_iter()
.map(|e| {
let outcome = match e.outcome {
Ok(a) => {
*self.admissions.entry(admission_name(a)).or_default() += 1;
admission_name(a).to_owned()
}
Err(r) => format!("refused:{}", refusal_name(r)),
};
json!({"group": e.group_id, "outcome": outcome})
})
.collect();
let objects = bundle.objects.len();
let link = bundle.group.link_hash();
self.groups.insert(group, HeldGroup { verified, objects: bundle.objects, link });
json!({"ok": true, "chain": chain, "released": released, "objects": objects})
}
/// The media object `object` of group `group`, its payload: `verified` when its hash is
/// the receipt's and the receipt's leaf reaches the signed root; `no_receipt` while the
/// group's receipts have not arrived; otherwise refused with the reason. A verified
/// object's result carries its content hash and its group receipt's link hash, which a
/// frame decoded from it names (WAI browser-sink §7.2).
pub fn object(&mut self, group: u64, object: u64, payload: &[u8]) -> Value {
let Some(h) = self.groups.get(&group) else { return json!({"ok": false, "reason": "no_receipt"}) };
let Some(r) = h.objects.get(object as usize) else { return self.refuse("no_object_receipt") };
match self.verifier.verify_payload(payload, r, &h.verified) {
Ok(()) => json!({"ok": true, "content_hash": hex::encode(r.content_hash), "group_receipt": hex::encode(h.link)}),
Err(_) => self.refuse("payload_mismatch"),
}
}
/// An object that did not arrive, accounted for by inclusion (jwp-receipts §4.4): its
/// receipt verifies against its group's signed root, without its payload.
pub fn inclusion(&mut self, group: u64, object: u64) -> Value {
let Some(h) = self.groups.get(&group) else { return json!({"ok": false, "reason": "no_receipt"}) };
let Some(r) = h.objects.get(object as usize) else { return json!({"ok": false, "reason": "no_object_receipt"}) };
match self.verifier.verify_object(r, &h.verified) {
Ok(()) => json!({"ok": true, "content_hash": hex::encode(r.content_hash)}),
Err(_) => self.refuse("inclusion_invalid"),
}
}
pub fn has_receipts(&self, group: u64) -> bool {
self.groups.contains_key(&group)
}
/// Forget groups below `group`: what a long-running page keeps is bounded.
pub fn forget_below(&mut self, group: u64) {
self.groups = self.groups.split_off(&group);
}
pub fn stats(&self) -> Value {
json!({
"groups_held": self.groups.len(),
"admissions": self.admissions,
"refusals": self.refusals,
"chain_held": self.chains.held_total(),
"anchors": self.chains.anchors(&self.track.publisher_key, self.track.track_alias).len(),
"receipts_bytes": self.sizes.json(),
"receipts_json_bytes": self.json_sizes.json(),
})
}
}
/// The subscriber's verifier of one media track, for a page.
#[wasm_bindgen]
pub struct TrackVerifier {
inner: Verifier,
}
#[wasm_bindgen]
impl TrackVerifier {
/// `publisher_key`: the 32-byte key the publisher signs with (the room principal's).
/// `receipt_alias`: the media track's `wai_receipt_alias`, from its catalog entry.
#[wasm_bindgen(constructor)]
pub fn new(publisher_key: &[u8], receipt_alias: u64) -> Result<TrackVerifier, JsError> {
let k: [u8; 32] = publisher_key.try_into().map_err(|_| JsError::new("a key is 32 bytes"))?;
Ok(TrackVerifier { inner: Verifier::new(k, receipt_alias).map_err(js)? })
}
pub fn receipts(&mut self, group: u64, payload: &[u8], now_ms: u64) -> String {
self.inner.receipts(group, payload, now_ms).to_string()
}
pub fn object(&mut self, group: u64, object: u64, payload: &[u8]) -> String {
self.inner.object(group, object, payload).to_string()
}
pub fn inclusion(&mut self, group: u64, object: u64) -> String {
self.inner.inclusion(group, object).to_string()
}
#[wasm_bindgen(js_name = hasReceipts)]
pub fn has_receipts(&self, group: u64) -> bool {
self.inner.has_receipts(group)
}
#[wasm_bindgen(js_name = forgetBelow)]
pub fn forget_below(&mut self, group: u64) {
self.inner.forget_below(group)
}
pub fn stats(&self) -> String {
self.inner.stats().to_string()
}
}
/// The header of a stream that begins with `bytes`, when they hold it: what a page reads
/// to route a stream before reading its objects.
pub fn header_of(bytes: &[u8]) -> Option<StreamHeader> {
let mut d = UniStreamDecoder::new(GroupOrder::Ascending, Limits::default());
d.push(bytes);
match d.next() {
Ok(Some(UniEvent::Header(h))) => Some(h),
_ => None,
}
}
#[cfg(feature = "e2ee")]
pub mod e2ee;
#[cfg(test)]
mod tests;