7e15664de9
A session whose SETUP presents a token whose rule says unlimited is exempt from max_session_ms and max_ingress_kbps: the room service's clock is a long-lived session of the deployment's own, and the session limit closed it every ten minutes (seen in the service scenario). joule-moq-room's relay-lines mark its clock's rule so. joule-moq-web: grantProofBy and clientSetupWithProof, so a test presents a page's grant with a proof by another key (the page's own path signs with the principal's key alone); the service scenario's teardown is bounded and its proxy ends upgraded sockets. Co-Authored-By: Claude Opus 5.5 <[email protected]>
907 lines
41 KiB
Rust
907 lines
41 KiB
Rust
//! **joule-moq-web**: `joule-moq`'s `draft-ietf-moq-transport-22` codec and its receipts,
|
|
//! for a browser.
|
|
//!
|
|
//! 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())
|
|
}
|
|
|
|
/// `moqt-22`: what a page offers in `protocols` and in the WebSocket subprotocol.
|
|
#[wasm_bindgen]
|
|
pub fn protocol() -> String {
|
|
PROTOCOL.into()
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------
|
|
// 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")?));
|
|
}
|
|
Ok(p)
|
|
}
|
|
|
|
/// 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));
|
|
}
|
|
ControlMessage::Setup(Setup { options: o }).encode().map_err(|e| e.to_string())
|
|
}
|
|
|
|
/// 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));
|
|
}
|
|
ControlMessage::Setup(Setup { options: o }).encode().map_err(|e| e.to_string())
|
|
}
|
|
|
|
/// [`setup_with_grant_bytes`] for a page: the challenge is the relay's `GRANT_CHALLENGE`
|
|
/// option (`0x3A47`), read from its SETUP.
|
|
#[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())
|
|
}
|
|
|
|
/// [`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
|
|
}
|
|
|
|
/// 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")),
|
|
};
|
|
m.encode().map_err(|e| e.to_string())
|
|
}
|
|
|
|
#[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()
|
|
}
|
|
|
|
/// Framed control messages (§9: Type, a 16-bit Length, the body) read from a request
|
|
/// stream as its bytes arrive. Each [`MessageStream::next`] gives the next whole message
|
|
/// as JSON (`draft22::json::control`), or nothing until more bytes come.
|
|
#[wasm_bindgen]
|
|
#[derive(Default)]
|
|
pub struct MessageStream {
|
|
buf: Vec<u8>,
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
pub fn next_message(&mut self) -> Result<Option<ControlMessage>, WireError> {
|
|
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),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[wasm_bindgen]
|
|
impl MessageStream {
|
|
#[wasm_bindgen(constructor)]
|
|
pub fn new() -> MessageStream {
|
|
MessageStream::default()
|
|
}
|
|
|
|
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> {
|
|
self.next_message().map(|m| m.map(|m| json::control(&m).to_string())).map_err(|e| JsError::new(&wire_error(&e)))
|
|
}
|
|
|
|
/// 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>,
|
|
}
|
|
|
|
#[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)
|
|
}
|
|
}
|
|
|
|
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,
|
|
},
|
|
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,
|
|
},
|
|
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() }
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
#[wasm_bindgen]
|
|
pub struct UniDecoder {
|
|
dec: UniStreamDecoder,
|
|
}
|
|
|
|
#[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 {
|
|
UniDecoder { dec: UniStreamDecoder::new(GroupOrder::Ascending, Limits { max_payload, ..Limits::default() }) }
|
|
}
|
|
|
|
/// 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) {
|
|
self.dec.set_group_order(if order == 2 { GroupOrder::Descending } else { GroupOrder::Ascending });
|
|
}
|
|
|
|
pub fn push(&mut self, bytes: &[u8]) {
|
|
self.dec.push(bytes);
|
|
}
|
|
|
|
#[allow(clippy::should_implement_trait)]
|
|
pub fn next(&mut self) -> Result<Option<UniItem>, JsError> {
|
|
self.dec.next().map(|e| e.map(item_of)).map_err(|e| JsError::new(&wire_error(&e)))
|
|
}
|
|
|
|
/// At the stream's FIN: a stream that ends inside an item is refused.
|
|
pub fn finish(&self) -> Result<(), JsError> {
|
|
self.dec.finish().map_err(|e| JsError::new(&wire_error(&e)))
|
|
}
|
|
}
|
|
|
|
/// 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]>,
|
|
}
|
|
|
|
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 }
|
|
}
|
|
|
|
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> {
|
|
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: true,
|
|
};
|
|
let props = 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()
|
|
}
|
|
|
|
/// 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) })
|
|
}
|
|
}
|
|
|
|
/// 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 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();
|
|
self.groups.insert(group, HeldGroup { verified, objects: bundle.objects });
|
|
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.
|
|
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}),
|
|
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(test)]
|
|
mod tests;
|