357 lines
16 KiB
Rust
357 lines
16 KiB
Rust
//! Checking objects against their receipts before they are forwarded, in the namespaces
|
|
//! the configuration names.
|
|
//!
|
|
//! Elsewhere the relay forwards a receipts track as it forwards any track, its objects
|
|
//! opaque (WAI `moq-streaming-format.md` §4.1; MOQT §7.8), and every subscriber verifies
|
|
//! for itself. In a namespace with a door rule the relay is a relay that reads the receipts
|
|
//! it forwards, and takes on the duties of WAI `jwp-receipts.md` §4.1:
|
|
//!
|
|
//! - each group receipt, from the bundle that group *g* of a receipts track carries, has
|
|
//! its Ed25519 signature verified (under the rule's trusted keys, or any key) before the
|
|
//! first object of media group *g* is fanned out. A group that does not verify is
|
|
//! dropped, its bundle and its objects, and a structured error is logged for the
|
|
//! upstream operator;
|
|
//! - each media object is checked against the receipt of its own position in the bundle:
|
|
//! the receipt names this group and Object ID (`moqt_mismatch` otherwise), its content
|
|
//! hash is the payload's, and its Merkle proof reaches the signed root. An object that
|
|
//! carries its receipt as a property must carry that same receipt;
|
|
//! - an object whose bundle has not arrived waits for it, up to the rule's `hold_ms`
|
|
//! (the publisher publishes the bundle no later than the group's first object), and is
|
|
//! refused after that;
|
|
//! - receipt bytes are forwarded unaltered; the relay re-signs nothing.
|
|
//!
|
|
//! The door reuses `joule-moq`'s verifier ([`ReceiptVerifier`]) and bundle reader.
|
|
//!
|
|
//! A rule names the keys it trusts, or (`trust_field`) the namespace field that names the
|
|
//! one key trusted there: a room's media namespaces are `[room, principal, kind]` (WAI
|
|
//! live-rooms §6.1), so a rule over `["*", "*", "video"]` with field 1 holds each
|
|
//! publisher's video to its own key, whoever joins.
|
|
|
|
use crate::config::DoorRule;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use crate::auth::pattern_matches;
|
|
use joule_moq::receipt::{ObjectReceipt, ReceiptVerifier, VerifiedGroup};
|
|
use joule_moq::{GroupId, ObjectId, TrackName, TrackNamespace};
|
|
use std::collections::{HashMap, VecDeque};
|
|
use std::sync::Mutex;
|
|
use std::time::Duration;
|
|
use tokio::sync::Notify;
|
|
|
|
/// How many groups of each track the door remembers.
|
|
const GROUPS_HELD: usize = 4096;
|
|
|
|
enum Trust {
|
|
/// The rule's keys, or any key.
|
|
Fixed(ReceiptVerifier),
|
|
/// The key a namespace field names, hex.
|
|
Field(usize),
|
|
}
|
|
|
|
struct Rule {
|
|
prefix: TrackNamespace,
|
|
trust: Trust,
|
|
hold: Duration,
|
|
}
|
|
|
|
/// A rule's verifier, or one made for a namespace.
|
|
enum Held<'a> {
|
|
Borrowed(&'a ReceiptVerifier),
|
|
Owned(ReceiptVerifier),
|
|
}
|
|
|
|
impl std::ops::Deref for Held<'_> {
|
|
type Target = ReceiptVerifier;
|
|
fn deref(&self) -> &ReceiptVerifier {
|
|
match self {
|
|
Held::Borrowed(v) => v,
|
|
Held::Owned(v) => v,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Rule {
|
|
/// The verifier for `ns`, or why there is none.
|
|
fn verifier(&self, ns: &TrackNamespace) -> Result<Held<'_>, String> {
|
|
match &self.trust {
|
|
Trust::Fixed(v) => Ok(Held::Borrowed(v)),
|
|
Trust::Field(i) => {
|
|
let key = ns
|
|
.0
|
|
.get(*i)
|
|
.and_then(|f| std::str::from_utf8(f).ok())
|
|
.filter(|f| f.len() == 64)
|
|
.and_then(|f| hex::decode(f).ok())
|
|
.and_then(|b| <[u8; 32]>::try_from(b).ok())
|
|
.and_then(|b| ed25519_dalek::VerifyingKey::from_bytes(&b).ok())
|
|
.ok_or_else(|| format!("namespace field {i} names no publisher key"))?;
|
|
Ok(Held::Owned(ReceiptVerifier::with_anchors(vec![key])))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
enum Entry {
|
|
Verified { group: VerifiedGroup, objects: HashMap<u64, ObjectReceipt> },
|
|
Refused(String),
|
|
}
|
|
|
|
type MediaKey = (TrackNamespace, TrackName);
|
|
|
|
#[derive(Default)]
|
|
struct Groups {
|
|
entries: HashMap<(MediaKey, u64), Entry>,
|
|
order: VecDeque<(MediaKey, u64)>,
|
|
}
|
|
|
|
pub struct Door {
|
|
rules: Vec<Rule>,
|
|
suffix: Vec<u8>,
|
|
groups: Mutex<Groups>,
|
|
arrived: Notify,
|
|
}
|
|
|
|
/// What the door decided about one object.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub enum Verdict {
|
|
/// No rule covers the namespace.
|
|
Unchecked,
|
|
/// The object's receipt verified: its signed figure and leaf class.
|
|
Pass { joules_micro: u64, class: Option<Vec<u8>> },
|
|
Refuse(String),
|
|
}
|
|
|
|
impl Door {
|
|
pub fn new(rules: &[DoorRule], suffix: &str) -> Result<Self, String> {
|
|
let mut out = Vec::new();
|
|
for r in rules {
|
|
let mut anchors = Vec::new();
|
|
for k in &r.trust {
|
|
let b: [u8; 32] = hex::decode(k).map_err(|e| format!("a door key: {e}"))?.try_into().map_err(|_| "a door key is not 32 bytes".to_string())?;
|
|
anchors.push(ed25519_dalek::VerifyingKey::from_bytes(&b).map_err(|e| format!("a door key: {e}"))?);
|
|
}
|
|
let trust = match r.trust_field {
|
|
Some(_) if !anchors.is_empty() => return Err("a door rule names keys or a trust_field, not both".into()),
|
|
Some(i) => Trust::Field(i),
|
|
None if anchors.is_empty() => Trust::Fixed(ReceiptVerifier::new()),
|
|
None => Trust::Fixed(ReceiptVerifier::with_anchors(anchors)),
|
|
};
|
|
out.push(Rule { prefix: TrackNamespace::from_strs(r.prefix.iter().map(String::as_str)), trust, hold: Duration::from_millis(r.hold_ms) });
|
|
}
|
|
Ok(Door { rules: out, suffix: suffix.as_bytes().to_vec(), groups: Mutex::new(Groups::default()), arrived: Notify::new() })
|
|
}
|
|
|
|
fn rule(&self, ns: &TrackNamespace) -> Option<&Rule> {
|
|
self.rules.iter().filter(|r| pattern_matches(&r.prefix, ns)).max_by_key(|r| r.prefix.0.len())
|
|
}
|
|
|
|
/// Whether `ns` is checked at the door.
|
|
pub fn checks(&self, ns: &TrackNamespace) -> bool {
|
|
self.rule(ns).is_some()
|
|
}
|
|
|
|
/// The verifier a stored group of `ns` is checked with (the DVR store's load): its rule's
|
|
/// trusted keys, or the key its `trust_field` names; any key whose signature verifies
|
|
/// where no rule covers `ns`.
|
|
pub fn verifier_for(&self, ns: &TrackNamespace) -> Result<ReceiptVerifier, String> {
|
|
let Some(rule) = self.rule(ns) else { return Ok(ReceiptVerifier::new()) };
|
|
let v = rule.verifier(ns)?;
|
|
Ok(ReceiptVerifier { trust_anchors: v.trust_anchors.clone(), require_known_publisher: v.require_known_publisher })
|
|
}
|
|
|
|
/// The media track a receipts track carries the receipts of, by the configured suffix.
|
|
pub fn media_of(&self, name: &TrackName) -> Option<TrackName> {
|
|
if self.suffix.is_empty() || name.0.len() <= self.suffix.len() || !name.0.ends_with(&self.suffix) {
|
|
return None;
|
|
}
|
|
Some(TrackName(name.0[..name.0.len() - self.suffix.len()].to_vec()))
|
|
}
|
|
|
|
/// The receipts track of a media track.
|
|
pub fn receipts_of(&self, name: &TrackName) -> TrackName {
|
|
let mut v = name.0.clone();
|
|
v.extend_from_slice(&self.suffix);
|
|
TrackName(v)
|
|
}
|
|
|
|
fn remember(&self, key: (MediaKey, u64), e: Entry) {
|
|
let mut g = self.groups.lock().unwrap();
|
|
if g.entries.insert(key.clone(), e).is_none() {
|
|
g.order.push_back(key);
|
|
while g.order.len() > GROUPS_HELD * self.rules.len().max(1) {
|
|
if let Some(old) = g.order.pop_front() {
|
|
g.entries.remove(&old);
|
|
}
|
|
}
|
|
}
|
|
drop(g);
|
|
self.arrived.notify_waiters();
|
|
}
|
|
|
|
/// Group `group` of the receipts track of `media` arrived, with this payload. `Ok` to
|
|
/// forward it, `Err` with why the group is dropped.
|
|
pub fn bundle(&self, ns: &TrackNamespace, media: &TrackName, group: u64, payload: &[u8]) -> Result<(), String> {
|
|
let Some(rule) = self.rule(ns) else { return Ok(()) };
|
|
let key = ((ns.clone(), media.clone()), group);
|
|
let checked = (|| {
|
|
let b = ReceiptBundle::from_payload(payload).map_err(|e| e.to_string())?;
|
|
if b.group.group_id != GroupId(group) {
|
|
return Err("moqt_mismatch: the group receipt names another group".to_string());
|
|
}
|
|
let verifier = rule.verifier(ns)?;
|
|
let vg = verifier.verify_group(&b.group).map_err(|e| e.to_string())?;
|
|
let mut objects = HashMap::new();
|
|
for o in b.objects {
|
|
o.check_moqt(b.group.track_alias, GroupId(group), o.object_id).map_err(|_| "moqt_mismatch: an object receipt of another track or group".to_string())?;
|
|
verifier.verify_object(&o, &vg).map_err(|e| format!("an object receipt: {e}"))?;
|
|
objects.insert(o.object_id.0, o);
|
|
}
|
|
Ok((vg, objects))
|
|
})();
|
|
match checked {
|
|
Ok((group, objects)) => {
|
|
self.remember(key, Entry::Verified { group, objects });
|
|
Ok(())
|
|
}
|
|
Err(why) => {
|
|
self.remember(key, Entry::Refused(why.clone()));
|
|
Err(why)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Check one media object. `property` is the receipt the object carries as its
|
|
/// JWP_OBJECT_RECEIPT property, if any.
|
|
pub async fn object(&self, ns: &TrackNamespace, media: &TrackName, group: u64, object: u64, payload: &[u8], property: Option<&ObjectReceipt>) -> Verdict {
|
|
let Some(rule) = self.rule(ns) else { return Verdict::Unchecked };
|
|
let key = ((ns.clone(), media.clone()), group);
|
|
let deadline = tokio::time::Instant::now() + rule.hold;
|
|
loop {
|
|
let notified = self.arrived.notified();
|
|
tokio::pin!(notified);
|
|
notified.as_mut().enable();
|
|
{
|
|
let g = self.groups.lock().unwrap();
|
|
match g.entries.get(&key) {
|
|
Some(Entry::Refused(why)) => return Verdict::Refuse(format!("its group was refused: {why}")),
|
|
Some(Entry::Verified { group: vg, objects }) => {
|
|
let Some(r) = objects.get(&object) else {
|
|
return Verdict::Refuse("no receipt for its position in the group".into());
|
|
};
|
|
if r.object_id != ObjectId(object) {
|
|
return Verdict::Refuse("moqt_mismatch".into());
|
|
}
|
|
if let Some(p) = property
|
|
&& p != r
|
|
{
|
|
return Verdict::Refuse("its receipt property is not the receipt its group signed".into());
|
|
}
|
|
let verifier = match rule.verifier(ns) {
|
|
Ok(v) => v,
|
|
Err(why) => return Verdict::Refuse(why),
|
|
};
|
|
return match verifier.verify_payload(payload, r, vg) {
|
|
Ok(()) => Verdict::Pass { joules_micro: r.joules_micro, class: r.energy_class.clone() },
|
|
Err(e) => Verdict::Refuse(format!("its payload does not verify: {e}")),
|
|
};
|
|
}
|
|
None => {}
|
|
}
|
|
}
|
|
if tokio::time::timeout_at(deadline, notified).await.is_err() {
|
|
return Verdict::Refuse("its group's receipts did not arrive in time".into());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use joule_moq::receipt::ReceiptBuilder;
|
|
use joule_moq::TrackAlias;
|
|
|
|
fn bundle(key: &ed25519_dalek::SigningKey, group: u64, payloads: &[&[u8]]) -> ReceiptBundle {
|
|
let mut b = ReceiptBuilder::new(TrackAlias(8), GroupId(group));
|
|
for p in payloads {
|
|
b.add_labelled_object(ObjectReceipt::hash_payload(p), 0, "o", "t", vec![0]).unwrap();
|
|
}
|
|
let objects = (0..payloads.len())
|
|
.map(|i| ObjectReceipt {
|
|
track_alias: TrackAlias(8),
|
|
group_id: GroupId(group),
|
|
object_id: ObjectId(i as u64),
|
|
content_hash: ObjectReceipt::hash_payload(payloads[i]),
|
|
joules_micro: 0,
|
|
origin: "o".into(),
|
|
merkle_proof: b.proof_for(i).unwrap(),
|
|
energy_class: Some(vec![0]),
|
|
})
|
|
.collect();
|
|
let group = b.finalize_labelled(key, "p", "t", None).unwrap();
|
|
ReceiptBundle { group, objects }
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_door_passes_what_its_group_signed() {
|
|
let key = ed25519_dalek::SigningKey::from_bytes(&[5; 32]);
|
|
let rule = DoorRule { prefix: vec!["live".into()], trust: vec![hex::encode(key.verifying_key().to_bytes())], trust_field: None, hold_ms: 100 };
|
|
let door = Door::new(&[rule], "_receipts").unwrap();
|
|
let ns = TrackNamespace::from_strs(["live", "a"]);
|
|
let media = TrackName::from_str("cam");
|
|
assert_eq!(door.media_of(&TrackName::from_str("cam_receipts")), Some(media.clone()));
|
|
assert_eq!(door.media_of(&media), None);
|
|
assert!(!door.checks(&TrackNamespace::from_strs(["other"])));
|
|
|
|
let b = bundle(&key, 3, &[b"x", b"y"]);
|
|
door.bundle(&ns, &media, 3, &b.to_payload().unwrap()).unwrap();
|
|
assert!(matches!(door.object(&ns, &media, 3, 1, b"y", None).await, Verdict::Pass { .. }));
|
|
assert!(matches!(door.object(&ns, &media, 3, 1, b"tampered", None).await, Verdict::Refuse(_)));
|
|
assert!(matches!(door.object(&ns, &media, 3, 0, b"x", Some(&b.objects[1])).await, Verdict::Refuse(_)), "a property that is not its receipt");
|
|
assert!(matches!(door.object(&ns, &media, 4, 0, b"x", None).await, Verdict::Refuse(_)), "no bundle within the hold");
|
|
|
|
// A bundle signed by a key the rule does not trust drops its group.
|
|
let stranger = ed25519_dalek::SigningKey::from_bytes(&[6; 32]);
|
|
let bad = bundle(&stranger, 5, &[b"z"]);
|
|
assert!(door.bundle(&ns, &media, 5, &bad.to_payload().unwrap()).is_err());
|
|
assert!(matches!(door.object(&ns, &media, 5, 0, b"z", None).await, Verdict::Refuse(_)));
|
|
// A bundle under the wrong MOQT group is refused before its signature is read.
|
|
assert!(door.bundle(&ns, &media, 9, &bundle(&key, 8, &[b"q"]).to_payload().unwrap()).is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_object_waits_for_its_bundle() {
|
|
let key = ed25519_dalek::SigningKey::from_bytes(&[5; 32]);
|
|
let rule = DoorRule { prefix: vec![], trust: vec![], trust_field: None, hold_ms: 2_000 };
|
|
let door = std::sync::Arc::new(Door::new(&[rule], "_receipts").unwrap());
|
|
let ns = TrackNamespace::from_strs(["any"]);
|
|
let media = TrackName::from_str("m");
|
|
let d2 = door.clone();
|
|
let (ns2, m2) = (ns.clone(), media.clone());
|
|
let waiter = tokio::spawn(async move { d2.object(&ns2, &m2, 0, 0, b"late", None).await });
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
door.bundle(&ns, &media, 0, &bundle(&key, 0, &[b"late"]).to_payload().unwrap()).unwrap();
|
|
let v = tokio::time::timeout(Duration::from_secs(5), waiter).await.unwrap().unwrap();
|
|
assert!(matches!(v, Verdict::Pass { .. }), "{v:?}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_trust_field_holds_each_namespace_to_the_key_it_names() {
|
|
let a = ed25519_dalek::SigningKey::from_bytes(&[5; 32]);
|
|
let b = ed25519_dalek::SigningKey::from_bytes(&[6; 32]);
|
|
let rule = DoorRule { prefix: vec!["*".into(), "*".into(), "video".into()], trust: vec![], trust_field: Some(1), hold_ms: 100 };
|
|
let door = Door::new(&[rule], "_receipts").unwrap();
|
|
let ns_a = TrackNamespace::from_strs(["room", &hex::encode(a.verifying_key().to_bytes()), "video"]);
|
|
let media = TrackName::from_str("0");
|
|
assert!(door.checks(&ns_a) && !door.checks(&TrackNamespace::from_strs(["room", "x", "events"])));
|
|
// a's groups verify in a's namespace; b's signature there drops its group.
|
|
door.bundle(&ns_a, &media, 0, &bundle(&a, 0, &[b"x"]).to_payload().unwrap()).unwrap();
|
|
assert!(matches!(door.object(&ns_a, &media, 0, 0, b"x", None).await, Verdict::Pass { .. }));
|
|
assert!(door.bundle(&ns_a, &media, 1, &bundle(&b, 1, &[b"y"]).to_payload().unwrap()).is_err());
|
|
// A namespace whose field is not a key: refused.
|
|
let odd = TrackNamespace::from_strs(["room", "not-a-key", "video"]);
|
|
assert!(door.bundle(&odd, &media, 0, &bundle(&a, 0, &[b"x"]).to_payload().unwrap()).is_err());
|
|
// Keys and a trust field together are not a rule.
|
|
let both = DoorRule { prefix: vec![], trust: vec![hex::encode(a.verifying_key().to_bytes())], trust_field: Some(1), hold_ms: 1 };
|
|
assert!(Door::new(&[both], "_receipts").is_err());
|
|
}
|
|
}
|