e3f79fa2be
Relay, by meaning: the DVR store's module text keeps both the minors' room (AGE) and the takedown (TKD) paragraphs; a session's object loop keeps CO4b's floor and audience-input filtering and TKD's takedown track feeding the store; tests/dvr.rs lists both rules. Shared lists kept both sides (workflow step and path, CHANGELOG, REGISTRY, SPEC §5, live-conformance index and provenance line, features, registry constants, lib.rs); export regenerated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1015 lines
57 KiB
Rust
1015 lines
57 KiB
Rust
//! The DVR store (WAI live-recording §8): a window of a room's tracks with their receipts,
|
|
//! on disk, served to FETCH below the live edge.
|
|
//!
|
|
//! - **The window.** A long publish crosses the window's byte budget: the oldest groups go,
|
|
//! whole, each media group with its receipts group, and the window holds what it says. A
|
|
//! window by age empties as its groups pass it.
|
|
//! - **FETCH inside, across and before the window**, ascending and descending: inside, every
|
|
//! object, each verifying against the fetched receipts; across, an End of Unknown Range
|
|
//! for what was evicted and every object after it, from the store and the cache together;
|
|
//! before, an End of Unknown Range alone.
|
|
//! - **A restart** keeps the store: every object is checked again, a corrupted one (a
|
|
//! flipped byte; bytes changed with their checks made again, so only the receipt finds
|
|
//! them) is dropped, counted and never served, and a write a crash cut short is removed.
|
|
//! - **Moderation.** A room key's redaction withholds its target from every DVR read and
|
|
//! erases it from disk; a deny withholds until an allow restores it; a record of an events
|
|
//! track is withheld from the object that carries it; a decision by someone who is not a
|
|
//! moderator changes nothing; and the decisions hold across a restart.
|
|
//! - **The seek power** (WAI live-rooms §10.6): a viewer whose grant lacks it is refused a
|
|
//! FETCH and a fill below the live edge (`seek_not_granted`) and served one at the live
|
|
//! edge; a viewer whose grant holds it is served the window.
|
|
//! - **Metering**: the relay signs `WMH1` version 2, with the store's figures.
|
|
//! - **A sealed room** (WAI confidentiality §9) is never kept: its first sealed object is
|
|
//! never written, what the store held of the room (its events and receipts, in the clear)
|
|
//! is deleted with its decisions, nothing more of it is written, and that holds across a
|
|
//! restart, while a room in the clear beside it keeps its window.
|
|
//! - **A room in which a minor publishes** (WAI age-assurance §8.4) is never kept either: once
|
|
//! a session proves a minor's publishing grant of the room, what the store held of it is
|
|
//! deleted, nothing more of it is written, and that holds across a restart, while another
|
|
//! room keeps its window; a minor's grant that publishes nothing changes nothing.
|
|
//! - **A signed takedown** (WAI takedown §10.1), riding its service's takedown track, makes the
|
|
//! store forget what it covers, in every room: an object by its content hash, a perceptual
|
|
//! match's candidate, and a track from a group on. None of it is served to a FETCH, from the
|
|
//! store or the cache, and its bytes leave the disk; an identical copy published later on
|
|
//! another track is never stored; the store signs a purge with the relay's key, once; a
|
|
//! takedown by a key the store does not trust, on its own track or on the service's,
|
|
//! changes nothing; and all of it holds across a restart.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::{Signer, SigningKey};
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft22::data::{FetchEntry, ObjectStatus, RangeEnd, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::{GroupOrder, Location, LocationFilter, Parameter, Parameters};
|
|
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder, ReceiptVerifier};
|
|
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackNamespace};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{Authority, DvrConfig, DvrWindow};
|
|
use joule_moq_relay::{RelayConfig, RelayServer};
|
|
use std::collections::BTreeMap;
|
|
use std::path::PathBuf;
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
use tokio::time::timeout;
|
|
|
|
const OBJECTS: u64 = 3;
|
|
const LEN: usize = 1000;
|
|
/// The cache's budget: about three groups with their receipts, so that what is older comes
|
|
/// from the store.
|
|
const CACHE: usize = 16 * 1024;
|
|
|
|
fn now_s() -> u64 {
|
|
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
|
|
}
|
|
|
|
/// A directory of its own for a store.
|
|
fn store_dir(name: &str) -> PathBuf {
|
|
let nanos = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
|
|
let d = std::env::temp_dir().join(format!("joule-moq-dvr-{name}-{}-{nanos}", std::process::id()));
|
|
std::fs::create_dir_all(&d).unwrap();
|
|
d
|
|
}
|
|
|
|
fn dvr_config(path: &std::path::Path, max_age_s: u64, max_bytes: u64, room_keys: Vec<String>) -> DvrConfig {
|
|
DvrConfig { path: path.to_path_buf(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s, max_bytes }], room_keys, takedown_keys: vec![] }
|
|
}
|
|
|
|
fn hexkey(k: &SigningKey) -> String {
|
|
hex::encode(k.verifying_key().to_bytes())
|
|
}
|
|
|
|
fn object_payload(tag: &str, g: u64, o: u64) -> Vec<u8> {
|
|
let mut v = format!("{tag}:{g}:{o}:").into_bytes();
|
|
v.resize(LEN, (g * 7 + o) as u8);
|
|
v
|
|
}
|
|
|
|
/// The receipts of a group of `payloads`, as group `g` of its receipts track carries them.
|
|
fn bundle_of(key: &SigningKey, receipt_alias: u64, g: u64, payloads: &[Vec<u8>]) -> ReceiptBundle {
|
|
let mut b = ReceiptBuilder::new(TrackAlias(receipt_alias), GroupId(g));
|
|
for p in payloads {
|
|
b.add_labelled_object(ObjectReceipt::hash_payload(p), 0, "test-publisher", "2026-10-19T00:00:00Z", vec![0]).unwrap();
|
|
}
|
|
let objects = payloads
|
|
.iter()
|
|
.enumerate()
|
|
.map(|(i, p)| ObjectReceipt {
|
|
track_alias: TrackAlias(receipt_alias),
|
|
group_id: GroupId(g),
|
|
object_id: ObjectId(i as u64),
|
|
content_hash: ObjectReceipt::hash_payload(p),
|
|
joules_micro: 0,
|
|
origin: "test-publisher".into(),
|
|
merkle_proof: b.proof_for(i).unwrap(),
|
|
energy_class: Some(vec![0]),
|
|
})
|
|
.collect();
|
|
let group = b.finalize_labelled(key, "test-publisher", "2026-10-19T00:00:01Z", None).unwrap();
|
|
ReceiptBundle { group, objects }
|
|
}
|
|
|
|
/// Publish group `g` of `t`: its receipts, then its objects on one subgroup stream.
|
|
async fn publish_multi(c: &Client, key: &SigningKey, t: &Tracks, g: u64, tag: &str) {
|
|
let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| object_payload(tag, g, o)).collect();
|
|
let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
|
|
let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
|
|
let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject::normal(o as u64, p)).collect();
|
|
c.send_subgroup(mh, &objects).await.expect("the objects");
|
|
}
|
|
|
|
fn room_tracks(room: [u8; 32], publisher: &SigningKey) -> Tracks {
|
|
Tracks::new(ns(&[&hex::encode(room), &hexkey(publisher), "video"]), "0", 10, 5, 128)
|
|
}
|
|
|
|
async fn flush(server: &RelayServer) {
|
|
let d = server.relay().dvr().expect("a DVR store").clone();
|
|
tokio::task::spawn_blocking(move || d.flush()).await.unwrap();
|
|
}
|
|
|
|
/// Wait, bounded, until the store has taken `n` groups: written, or found late (whole
|
|
/// after the window had moved past them, under load).
|
|
async fn written(server: &RelayServer, n: u64) {
|
|
let d = server.relay().dvr().expect("a DVR store").clone();
|
|
assert!(until(STEP, || d.counts().groups_written + d.counts().groups_late >= n).await, "{n} groups written: {:?}", d.counts());
|
|
flush(server).await;
|
|
}
|
|
|
|
fn range(start: (u64, u64), end_group: u64) -> Parameters {
|
|
Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(start.0, start.1), end_group_delta: end_group - start.0 }))
|
|
}
|
|
|
|
/// What a fetch stream carried: its objects by location, and its End of Range entries.
|
|
#[derive(Debug, Default)]
|
|
struct Fetched {
|
|
order: Vec<(u64, u64)>,
|
|
objects: BTreeMap<(u64, u64), Vec<u8>>,
|
|
unknown: Vec<Location>,
|
|
other_ends: Vec<(RangeEnd, Location)>,
|
|
}
|
|
|
|
fn fetched(entries: Vec<FetchEntry>) -> Fetched {
|
|
let mut f = Fetched::default();
|
|
for e in entries {
|
|
match e {
|
|
FetchEntry::Object(o) => {
|
|
f.order.push((o.group_id, o.object_id));
|
|
f.objects.insert((o.group_id, o.object_id), o.payload);
|
|
}
|
|
FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location } => f.unknown.push(location),
|
|
FetchEntry::EndOfRange { kind, location } => f.other_ends.push((kind, location)),
|
|
}
|
|
}
|
|
f
|
|
}
|
|
|
|
/// FETCH a range of the media track and of its receipts track; every media object fetched
|
|
/// verifies against the fetched receipts of its group under the publisher's key. Returns
|
|
/// the media fetched.
|
|
async fn fetch_verified(c: &Client, t: &Tracks, key: &SigningKey, params: Parameters) -> Fetched {
|
|
let (_, media) = c.fetch(t.ns.clone(), t.media.clone(), params.clone()).await.expect("FETCH the media track");
|
|
let (_, rec) = c.fetch(t.ns.clone(), t.receipts.clone(), params).await.expect("FETCH the receipts track");
|
|
let media = fetched(media);
|
|
let rec = fetched(rec);
|
|
let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
|
|
for ((g, o), p) in &media.objects {
|
|
let b = ReceiptBundle::from_payload(rec.objects.get(&(*g, 0)).unwrap_or_else(|| panic!("the receipts of group {g} come with it"))).unwrap();
|
|
let vg = v.verify_group(&b.group).expect("the group receipt verifies");
|
|
let r = b.objects.iter().find(|r| r.object_id == ObjectId(*o)).expect("a receipt for the object");
|
|
v.verify_payload(p, r, &vg).unwrap_or_else(|e| panic!("({g}, {o}) verifies against its receipt: {e}"));
|
|
}
|
|
media
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn the_window_and_fetch_inside_across_and_before_it() {
|
|
timeout(Duration::from_secs(180), window_run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn window_run() {
|
|
const GROUPS: u64 = 40;
|
|
let dir = store_dir("window");
|
|
let room = [0x51; 32];
|
|
let publisher = SigningKey::from_bytes(&[0x52; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
// About seven groups of media and receipts in the window; the cache holds fewer.
|
|
let max_bytes = 30_000;
|
|
let cfg = dvr_config(&dir, 900, max_bytes, vec![]);
|
|
let server = relay(|c| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
})
|
|
.await;
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_multi(&p, &publisher, &t, g, "w").await;
|
|
}
|
|
written(&server, 2 * GROUPS).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let media_key = (t.ns.clone(), t.media.clone());
|
|
let rec_key = (t.ns.clone(), t.receipts.clone());
|
|
let held = d.groups(&media_key);
|
|
let first = *held.first().expect("groups held");
|
|
assert!(first > 0, "the oldest groups were evicted");
|
|
assert_eq!(held, (first..GROUPS).collect::<Vec<_>>(), "the window is the newest groups, no gap");
|
|
assert_eq!(d.groups(&rec_key), held, "each media group goes with its receipts group");
|
|
let counts = d.counts();
|
|
assert!(counts.bytes_held <= max_bytes, "the window holds its budget: {counts:?}");
|
|
assert_eq!(counts.groups_evicted + counts.groups_late, 2 * first, "whole groups, media and receipts");
|
|
let all = GROUPS * (OBJECTS + 1);
|
|
assert!(counts.objects_written <= all && counts.objects_written + counts.groups_late * OBJECTS >= all, "every object, but a group the window had passed: {counts:?}");
|
|
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
// Inside the window: every object, verified, in the requested order.
|
|
let inside = fetch_verified(&v, &t, &publisher, range((first, 0), first + 2)).await;
|
|
assert_eq!(inside.objects.len() as u64, 3 * OBJECTS);
|
|
assert!(inside.unknown.is_empty() && inside.other_ends.is_empty());
|
|
for ((g, o), p) in &inside.objects {
|
|
assert_eq!(p, &object_payload("w", *g, *o));
|
|
}
|
|
let desc = fetch_verified(&v, &t, &publisher, range((first, 0), first + 2).with(Parameter::GroupOrder(GroupOrder::Descending))).await;
|
|
let groups: Vec<u64> = desc.order.iter().map(|x| x.0).collect();
|
|
assert_eq!(groups, [first + 2, first + 2, first + 2, first + 1, first + 1, first + 1, first, first, first], "descending groups, objects ascending");
|
|
assert!(desc.order.windows(2).all(|w| w[0].0 != w[1].0 || w[0].1 < w[1].1));
|
|
|
|
// Across: what was evicted is unknown, and the rest comes, from the store and the cache.
|
|
let across = fetch_verified(&v, &t, &publisher, Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))).await;
|
|
assert_eq!(across.unknown, vec![Location::new(first - 1, u64::MAX)], "End of Unknown Range through the last evicted group");
|
|
assert!(across.other_ends.is_empty());
|
|
assert_eq!(across.objects.len() as u64, (GROUPS - first) * OBJECTS, "every object the window holds and the cache's");
|
|
assert_eq!(across.order.first(), Some(&(first, 0)));
|
|
assert_eq!(across.order.last(), Some(&(GROUPS - 1, OBJECTS - 1)));
|
|
assert!(server.relay().stats().fetch_served_from_dvr >= 2);
|
|
|
|
// Before: nothing but the End of Unknown Range.
|
|
assert!(first > 3);
|
|
let (ok, before) = v.fetch(t.ns.clone(), t.media.clone(), range((0, 0), 2)).await.expect("a FETCH before the window");
|
|
assert_eq!(ok.end_location, Location::new(2, u64::MAX));
|
|
let before = fetched(before);
|
|
assert!(before.objects.is_empty());
|
|
assert_eq!(before.unknown, vec![Location::new(2, u64::MAX)]);
|
|
|
|
// The record of the room is version 2, with the store's figures.
|
|
v.close(0, "done");
|
|
p.close(0, "done");
|
|
assert!(until(2 * STEP, || server.relay().summaries().len() >= 2).await, "both sessions ended: {:?}; open: {:?}", server.relay().stats(), server.relay().session_diagnostics());
|
|
let rec = server.relay().hop_record(room, 0);
|
|
let bytes = rec.to_bytes();
|
|
assert_eq!(&bytes[..5], b"WMH1\x02", "version 2");
|
|
rec.verify().expect("the record verifies under the relay's key");
|
|
let back = joule_moq_relay::metering::HopRecord::decode(&bytes, &rec.ledger_bytes()).expect("it reads back");
|
|
assert_eq!(back, rec);
|
|
let dv = rec.dvr.expect("the DVR figures");
|
|
let served: u64 = server.relay().summaries().iter().flat_map(|s| s.summary.tally.subscriber.tracks.values().map(|x| x.dvr_objects)).sum();
|
|
assert_eq!(dv.objects_out, served);
|
|
assert!(dv.objects_out >= 2 * 3 * OBJECTS, "the inside fetches came from the store: {dv:?}");
|
|
let out: u64 = rec.entries.iter().filter(|e| e.dir == 2).map(|e| e.objects).sum();
|
|
let inn: u64 = rec.entries.iter().filter(|e| e.dir == 1).map(|e| e.objects).sum();
|
|
assert!(dv.objects_out <= out && dv.objects_stored <= inn, "{dv:?}, out {out}, in {inn}");
|
|
assert_eq!(dv.objects_stored, counts.objects_written, "every object of the room written in the span");
|
|
assert!(dv.byte_s > 0 && dv.dropped == 0);
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_window_by_age_empties_as_its_groups_pass_it() {
|
|
timeout(Duration::from_secs(60), async {
|
|
let dir = store_dir("age");
|
|
let room = [0x53; 32];
|
|
let publisher = SigningKey::from_bytes(&[0x54; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
let cfg = dvr_config(&dir, 2, 1 << 30, vec![]);
|
|
let server = relay(|c| {
|
|
// About two groups: what aged out of the store is out of the cache too.
|
|
c.cache.max_bytes = 8 * 1024;
|
|
c.dvr = Some(cfg.clone());
|
|
})
|
|
.await;
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..3 {
|
|
publish_multi(&p, &publisher, &t, g, "a").await;
|
|
}
|
|
written(&server, 6).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let (key, rkey) = ((t.ns.clone(), t.media.clone()), (t.ns.clone(), t.receipts.clone()));
|
|
assert!(until(Duration::from_secs(10), || d.groups(&key).is_empty() && d.groups(&rkey).is_empty()).await, "past max_age_s every group goes");
|
|
assert_eq!(d.counts().groups_evicted, 6);
|
|
// The next group, fetched at once: what aged out is unknown, and the group is served.
|
|
publish_multi(&p, &publisher, &t, 3, "a").await;
|
|
// The relay has taken group 3 into both tracks before the fetches: taking it may
|
|
// evict a pair from the cache, and the media and receipts FETCHes are two requests,
|
|
// so a pair evicted between them would leave media without receipts.
|
|
let relay = server.relay().clone();
|
|
assert!(
|
|
until(STEP, || relay.largest(&key) == Some(Location::new(3, OBJECTS - 1)) && relay.largest(&rkey) == Some(Location::new(3, 0))).await,
|
|
"the relay took group 3"
|
|
);
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, range((0, 0), 3)).await;
|
|
// What aged out is unknown through the last group neither the store nor the cache
|
|
// holds; every group after it comes whole.
|
|
assert_eq!(f.unknown.len(), 1);
|
|
let k = f.unknown[0];
|
|
assert!(k.object == u64::MAX && k.group < 3, "{k:?}");
|
|
let want: Vec<(u64, u64)> = (k.group + 1..=3).flat_map(|g| (0..OBJECTS).map(move |o| (g, o))).collect();
|
|
assert_eq!(f.objects.keys().copied().collect::<Vec<_>>(), want);
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// Rewrite one object of the group record whose frame is at `frame_at` in a segment, in
|
|
/// place: `f` changes its payload (same length); with `fix`, its digest and check are made
|
|
/// again so that only its receipt finds it.
|
|
fn tamper(path: &std::path::Path, frame_at: u64, object: u64, fix: bool, f: impl Fn(&mut [u8])) {
|
|
let mut b = std::fs::read(path).unwrap();
|
|
let r = frame_at as usize + 5;
|
|
let count = u32::from_le_bytes(b[r + 53..r + 57].try_into().unwrap()) as u64;
|
|
let mut p = r + 93;
|
|
for i in 0..count {
|
|
let props_len = u32::from_le_bytes(b[p + 18..p + 22].try_into().unwrap()) as usize;
|
|
let payload_len = u32::from_le_bytes(b[p + 22..p + 26].try_into().unwrap()) as usize;
|
|
let at = p + 90 + props_len;
|
|
if i == object {
|
|
f(&mut b[at..at + payload_len]);
|
|
if fix {
|
|
let digest = ObjectReceipt::hash_payload(&b[at..at + payload_len]);
|
|
b[p + 26..p + 58].copy_from_slice(&digest);
|
|
let mut c = ring::digest::Context::new(&ring::digest::SHA256);
|
|
c.update(&b[p..p + 58]);
|
|
c.update(&b[p + 90..at + payload_len]);
|
|
let check = c.finish();
|
|
b[p + 58..p + 90].copy_from_slice(check.as_ref());
|
|
}
|
|
std::fs::write(path, &b).unwrap();
|
|
return;
|
|
}
|
|
p = at + payload_len;
|
|
}
|
|
panic!("no object {object}");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_restart_keeps_the_store_and_drops_what_does_not_verify() {
|
|
timeout(Duration::from_secs(120), async {
|
|
const GROUPS: u64 = 10;
|
|
let dir = store_dir("restart");
|
|
let room = [0x55; 32];
|
|
let publisher = SigningKey::from_bytes(&[0x56; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
|
|
let config = |c: &mut RelayConfig| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
};
|
|
let server = relay(config).await;
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_multi(&p, &publisher, &t, g, "r").await;
|
|
}
|
|
written(&server, 2 * GROUPS).await;
|
|
let key = (t.ns.clone(), t.media.clone());
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let ((g3, at3), (g5, at5), (g9, _)) = (d.group_location(&key, 3).unwrap(), d.group_location(&key, 5).unwrap(), d.group_location(&key, 9).unwrap());
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
|
|
// A flipped byte in (3, 1); (5, 0) rewritten with its checks made again; and a frame a
|
|
// crash cut short at the end of a segment.
|
|
tamper(&g3, at3, 1, false, |p| p[10] ^= 0x20);
|
|
tamper(&g5, at5, 0, true, |p| p.reverse());
|
|
let whole = std::fs::metadata(&g9).unwrap().len();
|
|
{
|
|
use std::io::Write;
|
|
let mut f = std::fs::OpenOptions::new().append(true).open(&g9).unwrap();
|
|
f.write_all(&1000u32.to_le_bytes()).unwrap();
|
|
f.write_all(&[1, b'W', b'D', b'G', b'1']).unwrap();
|
|
}
|
|
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let c = d.counts();
|
|
assert_eq!((c.objects_dropped, c.torn_writes), (2, 1), "{c:?}");
|
|
assert_eq!(std::fs::metadata(&g9).unwrap().len(), whole, "the cut-short frame is cut off");
|
|
assert_eq!(d.groups(&key), (0..GROUPS).collect::<Vec<_>>());
|
|
// No publisher now: the store answers alone.
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }))).await;
|
|
// An End of Range moves to a later group than the entry before it (MOQT §11.4.1.2),
|
|
// so (3, 1)'s marks group 3 from its start: (3, 0) goes with it.
|
|
assert_eq!(f.unknown, vec![Location::new(3, 1), Location::new(5, 0)], "each in its place as unknown");
|
|
assert!(!f.objects.contains_key(&(3, 1)) && !f.objects.contains_key(&(5, 0)), "never served");
|
|
assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 3, "every object but the two dropped and the one before (3, 1)");
|
|
assert!(f.objects.contains_key(&(3, 2)) && f.objects.contains_key(&(5, 1)), "the objects after them follow");
|
|
assert_eq!(server.relay().largest(&key), Some(Location::new(GROUPS - 1, OBJECTS - 1)));
|
|
// The dropped objects are out of their files too: a second restart finds nothing to drop.
|
|
server.shutdown().await;
|
|
let server = relay(config).await;
|
|
assert_eq!(server.relay().dvr().unwrap().counts().objects_dropped, 0);
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, range((3, 0), 5)).await;
|
|
assert_eq!(f.unknown, vec![Location::new(3, 1), Location::new(5, 0)]);
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// A record of live-events §3.1, without an anchor, signed by `key`.
|
|
fn event(key: &SigningKey, room: [u8; 32], seq: u64, t: u64, kind: u8, payload: &[u8]) -> Vec<u8> {
|
|
let mut b = b"WEV1\x01".to_vec();
|
|
b.extend_from_slice(&room);
|
|
b.extend_from_slice(&key.verifying_key().to_bytes());
|
|
b.extend_from_slice(&seq.to_le_bytes());
|
|
b.extend_from_slice(&[0; 32]);
|
|
b.extend_from_slice(&t.to_le_bytes());
|
|
b.push(0);
|
|
b.push(kind);
|
|
b.extend_from_slice(&(payload.len() as u16).to_le_bytes());
|
|
b.extend_from_slice(payload);
|
|
let mut m = b"wai:live-event\x01".to_vec();
|
|
m.extend_from_slice(&b);
|
|
b.extend_from_slice(&key.sign(&m).to_bytes());
|
|
b
|
|
}
|
|
|
|
fn event_digest(rec: &[u8]) -> [u8; 32] {
|
|
ring::digest::digest(&ring::digest::SHA256, &rec[..rec.len() - 64]).as_ref().try_into().unwrap()
|
|
}
|
|
|
|
fn decision(key: &SigningKey, room: [u8; 32], seq: u64, verdict: u8, target: [u8; 32]) -> Vec<u8> {
|
|
let mut p = vec![verdict];
|
|
p.extend_from_slice(&target);
|
|
p.extend_from_slice(&1u16.to_le_bytes());
|
|
event(key, room, seq, 1_000 + seq, 0x0B, &p)
|
|
}
|
|
|
|
fn msf(records: &[&[u8]]) -> Vec<u8> {
|
|
use base64::Engine;
|
|
let e: Vec<String> = records.iter().map(|r| format!("{{\"t\":1,\"data\":{{\"wev1\":\"{}\"}}}}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r))).collect();
|
|
format!("[{}]", e.join(",")).into_bytes()
|
|
}
|
|
|
|
/// A principal's events track: PUBLISH it, and send objects of group 0 one stream each.
|
|
struct EventsTrack {
|
|
ns: TrackNamespace,
|
|
alias: u64,
|
|
}
|
|
|
|
impl EventsTrack {
|
|
async fn open(c: &Client, room: [u8; 32], who: &SigningKey, alias: u64) -> (EventsTrack, (joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx)) {
|
|
let ns = ns(&[&hex::encode(room), &hexkey(who), "events"]);
|
|
let held = c.publish(ns.clone(), name("0"), alias, Parameters::new(), track_props(16, None)).await.expect("PUBLISH an events track");
|
|
(EventsTrack { ns, alias }, held)
|
|
}
|
|
|
|
/// Group `g`'s objects: the join set, then one object a record.
|
|
async fn group(&self, c: &Client, g: u64, records: &[Vec<u8>]) {
|
|
let h = SubgroupHeader { track_alias: self.alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
|
|
let mut objects = vec![SubgroupObject::normal(0, b"[]".to_vec())];
|
|
for (i, r) in records.iter().enumerate() {
|
|
objects.push(SubgroupObject::normal(i as u64 + 1, msf(&[r])));
|
|
}
|
|
c.send_subgroup(h, &objects).await.expect("the events");
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn decisions_withhold_from_the_store_and_hold_across_a_restart() {
|
|
timeout(Duration::from_secs(120), async {
|
|
const GROUPS: u64 = 6;
|
|
let dir = store_dir("moderation");
|
|
let room = [0x57; 32];
|
|
let publisher = SigningKey::from_bytes(&[0x58; 32]);
|
|
let moderator = SigningKey::from_bytes(&[0x59; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
let cfg = dvr_config(&dir, 900, 1 << 30, vec![hexkey(&moderator)]);
|
|
let config = |c: &mut RelayConfig| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
};
|
|
let server = relay(config).await;
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_multi(&p, &publisher, &t, g, "m").await;
|
|
}
|
|
// The publisher's events: two chat messages.
|
|
let (pe, _pe_held) = EventsTrack::open(&p, room, &publisher, 40).await;
|
|
let chat1 = event(&publisher, room, 0, 10, 0x03, b"\x02\x00hi");
|
|
let chat2 = event(&publisher, room, 1, 11, 0x03, b"\x03\x00bye");
|
|
pe.group(&p, 0, &[chat1.clone(), chat2.clone()]).await;
|
|
written(&server, 2 * GROUPS + 1).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
|
|
let h = |g: u64, o: u64| ObjectReceipt::hash_payload(&object_payload("m", g, o));
|
|
// The moderator (a room key): redact (2, 1), deny (3, 0), redact the first chat. The
|
|
// publisher, no moderator, redacts (4, 0): nothing comes of it.
|
|
let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (me, _me_held) = EventsTrack::open(&m, room, &moderator, 50).await;
|
|
me.group(&m, 0, &[decision(&moderator, room, 0, 3, h(2, 1)), decision(&moderator, room, 1, 2, h(3, 0)), decision(&moderator, room, 2, 3, event_digest(&chat1))]).await;
|
|
pe.group(&p, 1, &[decision(&publisher, room, 2, 3, h(4, 0))]).await;
|
|
assert!(until(STEP, || d.counts().decisions >= 3 && d.counts().objects_redacted >= 2).await, "{:?}", d.counts());
|
|
assert_eq!(d.counts().decisions, 3, "the publisher's decision is not honoured");
|
|
|
|
let all = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, all()).await;
|
|
assert_eq!(f.unknown, vec![Location::new(2, 1), Location::new(3, 0)], "withheld, each in its place");
|
|
assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 3, "and (2, 0), which an End of Range through (2, 1) covers");
|
|
assert!(f.objects.contains_key(&(2, 2)) && f.objects.contains_key(&(3, 1)));
|
|
assert!(f.objects.contains_key(&(4, 0)), "a decision by no moderator withholds nothing");
|
|
// The redacted object's bytes are gone from every file of its track; the denied one's stay.
|
|
let on_disk = |key: &(TrackNamespace, joule_moq::TrackName), bytes: &[u8]| -> bool {
|
|
let dir = d.track_dir(key).unwrap();
|
|
std::fs::read_dir(dir).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(bytes.len()).any(|w| w == bytes))
|
|
};
|
|
let media = (t.ns.clone(), t.media.clone());
|
|
assert!(!on_disk(&media, &object_payload("m", 2, 1)), "redacted: erased from disk");
|
|
assert!(on_disk(&media, &object_payload("m", 3, 0)), "denied: kept");
|
|
// The redacted chat is erased from the events track's files; the other stays.
|
|
let ev_key = (pe.ns.clone(), name("0"));
|
|
assert!(!on_disk(&ev_key, &msf(&[&chat1])), "redacted: erased from disk");
|
|
assert!(on_disk(&ev_key, &msf(&[&chat2])));
|
|
// The publisher's events: the redacted chat is withheld, the other is served as sent.
|
|
let (_, ev) = v.fetch(pe.ns.clone(), name("0"), all()).await.expect("FETCH the events track");
|
|
let ev = fetched(ev);
|
|
assert_eq!(ev.unknown, vec![Location::new(0, 1)], "the object holding the redacted record");
|
|
assert_eq!(ev.objects.get(&(0, 2)), Some(&msf(&[&chat2])));
|
|
|
|
// An allow supersedes the deny: (3, 0) is served again.
|
|
me.group(&m, 1, &[decision(&moderator, room, 3, 1, h(3, 0))]).await;
|
|
assert!(until(STEP, || d.counts().decisions >= 4).await);
|
|
let f = fetch_verified(&v, &t, &publisher, all()).await;
|
|
assert_eq!(f.unknown, vec![Location::new(2, 1)]);
|
|
assert!(f.objects.contains_key(&(3, 0)));
|
|
assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS - 2);
|
|
|
|
// Across a restart the decisions hold.
|
|
for c in [&p, &m, &v] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let server = relay(config).await;
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, all()).await;
|
|
assert_eq!(f.unknown, vec![Location::new(2, 1)]);
|
|
let (_, ev) = v.fetch(pe.ns.clone(), name("0"), all()).await.expect("FETCH the events track");
|
|
assert_eq!(fetched(ev).unknown, vec![Location::new(0, 1)]);
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// Publish group `g` of `t` sealed: its receipts, then objects each carrying a secure-objects
|
|
/// Key ID in its Immutable Properties (WAI confidentiality §4.3).
|
|
async fn publish_sealed(c: &Client, key: &SigningKey, t: &Tracks, g: u64) {
|
|
use joule_moq::secure_objects::{immutable_properties, object_properties};
|
|
let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| object_payload("sealed", g, o)).collect();
|
|
let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
|
|
let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
|
|
let props = object_properties(&immutable_properties(7, &[]), &[]);
|
|
let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: true, end_of_group: true, first_object: true };
|
|
let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject { object_id: o as u64, properties: props.clone(), status: ObjectStatus::Normal, payload: p }).collect();
|
|
c.send_subgroup(mh, &objects).await.expect("the sealed objects");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_sealed_room_is_never_kept() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let dir = store_dir("sealed");
|
|
let (sealed, clear) = ([0x5a; 32], [0x5b; 32]);
|
|
let publisher = SigningKey::from_bytes(&[0x5c; 32]);
|
|
let moderator = SigningKey::from_bytes(&[0x5d; 32]);
|
|
let ts = room_tracks(sealed, &publisher);
|
|
let tc = Tracks::new(ns(&[&hex::encode(clear), &hexkey(&publisher), "video"]), "0", 20, 6, 128);
|
|
let cfg = dvr_config(&dir, 900, 1 << 30, vec![hexkey(&moderator)]);
|
|
let config = |c: &mut RelayConfig| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
};
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _ps = publish_tracks(&p, &ts, None).await;
|
|
let _pc = publish_tracks(&p, &tc, None).await;
|
|
// Before the sealed room's first media group: its events, in the clear, and a
|
|
// moderator's decision in each room.
|
|
let (pe, _pe_held) = EventsTrack::open(&p, sealed, &publisher, 40).await;
|
|
let chat = event(&publisher, sealed, 0, 10, 0x03, b"\x02\x00hi");
|
|
pe.group(&p, 0, &[chat]).await;
|
|
let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (ms, _ms_held) = EventsTrack::open(&m, sealed, &moderator, 50).await;
|
|
let (mc, _mc_held) = EventsTrack::open(&m, clear, &moderator, 60).await;
|
|
let (ds, dc) = (decision(&moderator, sealed, 0, 2, [1; 32]), decision(&moderator, clear, 0, 2, [2; 32]));
|
|
ms.group(&m, 0, std::slice::from_ref(&ds)).await;
|
|
mc.group(&m, 0, std::slice::from_ref(&dc)).await;
|
|
for g in 0..2 {
|
|
publish_multi(&p, &publisher, &tc, g, "c").await;
|
|
}
|
|
written(&server, 7).await;
|
|
assert!(until(STEP, || d.counts().decisions >= 2).await, "{:?}", d.counts());
|
|
let ev_key = (pe.ns.clone(), name("0"));
|
|
let ev_dir = d.track_dir(&ev_key).expect("the sealed room's events, kept so far");
|
|
assert!(ev_dir.exists() && d.groups(&ev_key) == vec![0]);
|
|
let log = dir.join("decisions.wdl");
|
|
let holds = |path: &std::path::Path, bytes: &[u8]| std::fs::read(path).unwrap_or_default().windows(bytes.len()).any(|w| w == bytes);
|
|
assert!(holds(&log, &ds) && holds(&log, &dc));
|
|
|
|
// The first sealed group: never written, and the room is forgotten.
|
|
publish_sealed(&p, &publisher, &ts, 0).await;
|
|
assert!(until(STEP, || d.counts().rooms_sealed == 1).await, "{:?}", d.counts());
|
|
flush(&server).await;
|
|
let room_keys = [(ts.ns.clone(), ts.media.clone()), (ts.ns.clone(), ts.receipts.clone()), ev_key.clone(), (ms.ns.clone(), name("0"))];
|
|
for k in &room_keys {
|
|
assert!(d.groups(k).is_empty() && d.track_dir(k).is_none(), "{k:?} is kept");
|
|
}
|
|
assert!(!ev_dir.exists(), "the sealed room's events are deleted from disk");
|
|
assert!(!holds(&log, &ds), "its decision leaves the log");
|
|
assert!(holds(&log, &dc), "the clear room's stays");
|
|
assert_eq!(std::fs::metadata(dir.join("sealed.wdl")).unwrap().len(), 64);
|
|
let clear_media = (tc.ns.clone(), tc.media.clone());
|
|
assert_eq!(d.groups(&clear_media), vec![0, 1], "the clear room keeps its window");
|
|
|
|
// Nothing more of it is written: its events and its next sealed group, then a group
|
|
// of the clear room, which is.
|
|
pe.group(&p, 1, &[event(&publisher, sealed, 1, 11, 0x03, b"\x03\x00bye")]).await;
|
|
publish_sealed(&p, &publisher, &ts, 1).await;
|
|
assert!(until(STEP, || server.relay().largest(&(ts.ns.clone(), ts.media.clone())) == Some(Location::new(1, OBJECTS - 1))).await);
|
|
publish_multi(&p, &publisher, &tc, 2, "c").await;
|
|
assert!(until(STEP, || d.groups(&clear_media).contains(&2)).await);
|
|
flush(&server).await;
|
|
for k in &room_keys {
|
|
assert!(d.groups(k).is_empty(), "{k:?} is written after the room was sealed");
|
|
}
|
|
|
|
// Across a restart: the room is still sealed, nothing of it is loaded or written, and
|
|
// only the clear room's decision is honoured again.
|
|
for c in [&p, &m] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
assert_eq!((d.counts().rooms_sealed, d.counts().decisions), (1, 1), "{:?}", d.counts());
|
|
assert_eq!(d.groups(&clear_media), vec![0, 1, 2]);
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let (pe, _pe_held) = EventsTrack::open(&p, sealed, &publisher, 40).await;
|
|
pe.group(&p, 2, &[event(&publisher, sealed, 2, 12, 0x03, b"\x02\x00ok")]).await;
|
|
let _pc = publish_tracks(&p, &tc, None).await;
|
|
publish_multi(&p, &publisher, &tc, 3, "c").await;
|
|
assert!(until(STEP, || d.groups(&clear_media).contains(&3)).await);
|
|
flush(&server).await;
|
|
assert!(d.groups(&ev_key).is_empty() && d.track_dir(&ev_key).is_none());
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// Publish group `g` of `t` with these payloads: its receipts, then its objects.
|
|
async fn publish_payloads(c: &Client, key: &SigningKey, t: &Tracks, g: u64, payloads: Vec<Vec<u8>>) {
|
|
let bundle = bundle_of(key, t.receipt_alias, g, &payloads);
|
|
let rh = SubgroupHeader { track_alias: t.receipts_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
c.send_subgroup(rh, &[SubgroupObject::normal(0, bundle.to_payload().unwrap())]).await.expect("the receipts");
|
|
let mh = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
|
|
let objects: Vec<SubgroupObject> = payloads.into_iter().enumerate().map(|(o, p)| SubgroupObject::normal(o as u64, p)).collect();
|
|
c.send_subgroup(mh, &objects).await.expect("the objects");
|
|
}
|
|
|
|
/// A takedown as WAI takedown §5.3 lays it out, signed by `by`, naming `exact` and `matched`.
|
|
fn takedown_record(by: &SigningKey, exact: Vec<serde_json::Value>, matched: Vec<serde_json::Value>) -> (Vec<u8>, [u8; 32]) {
|
|
let mut v = serde_json::json!({
|
|
"kind": "wai-takedown", "policy": "00".repeat(32), "notice": null, "clock": null, "clock_start": null, "reply_at": null,
|
|
"targets": [exact[0].clone()], "verdict": "remove", "exact": exact, "matched": matched, "ground": "illegal", "category": 3,
|
|
"automated": {"detection": "no", "decision": "not"}, "explanation": "The item is unlawful.", "facts": "A notice named it.",
|
|
"decided_at": "2026-10-21T08:00:00Z", "signer_pubkey": hex::encode(by.verifying_key().to_bytes()),
|
|
});
|
|
let sig = by.sign(&joule_moq_relay::takedown::signing_bytes(&v).unwrap());
|
|
v["sig"] = serde_json::json!(hex::encode(sig.to_bytes()));
|
|
let link = joule_moq_relay::takedown::link_hash(&v).unwrap();
|
|
(serde_json::to_vec(&v).unwrap(), link)
|
|
}
|
|
|
|
/// Publish `records` on the takedown track of `owner`, `[owner, takedown]`, `0`: one group,
|
|
/// one object a record.
|
|
async fn publish_takedowns(c: &Client, owner: &SigningKey, alias: u64, records: &[Vec<u8>]) -> (joule_moq_relay::wire::ReqTx, joule_moq_relay::wire::ReqRx) {
|
|
let held = c.publish(ns(&[&hexkey(owner), "takedown"]), name("0"), alias, Parameters::new(), track_props(16, None)).await.expect("PUBLISH a takedown track");
|
|
let h = SubgroupHeader { track_alias: alias, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(16), properties: false, end_of_group: true, first_object: true };
|
|
let objects: Vec<SubgroupObject> = records.iter().enumerate().map(|(i, r)| SubgroupObject::normal(i as u64, r.clone())).collect();
|
|
c.send_subgroup(h, &objects).await.expect("the takedowns");
|
|
held
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_signed_takedown_forgets_what_it_covers_and_holds_across_a_restart() {
|
|
timeout(Duration::from_secs(150), async {
|
|
const GROUPS: u64 = 5;
|
|
let dir = store_dir("takedown");
|
|
let key_dir = store_dir("takedown-key");
|
|
let key_file = key_dir.join("relay.key");
|
|
std::fs::write(&key_file, [0x67u8; 32]).unwrap();
|
|
let (room, room2) = ([0x63; 32], [0x68; 32]);
|
|
let publisher = SigningKey::from_bytes(&[0x64; 32]);
|
|
let service = SigningKey::from_bytes(&[0x65; 32]);
|
|
let stranger = SigningKey::from_bytes(&[0x66; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
let t2 = Tracks::new(ns(&[&hex::encode(room2), &hexkey(&publisher), "video"]), "0", 30, 7, 128);
|
|
let mut cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
|
|
cfg.takedown_keys = vec![hexkey(&service)];
|
|
let config = |c: &mut RelayConfig| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
c.summary.key_path = Some(key_file.clone());
|
|
};
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_multi(&p, &publisher, &t, g, "td").await;
|
|
}
|
|
written(&server, 2 * GROUPS).await;
|
|
|
|
let h = |g: u64, o: u64| ObjectReceipt::hash_payload(&object_payload("td", g, o));
|
|
let object = |x: [u8; 32]| serde_json::json!({"type": "object", "content_hash": hex::encode(x)});
|
|
let media = (t.ns.clone(), t.media.clone());
|
|
let track = joule_moq_relay::dvr::room_track_digest(&media);
|
|
let exact = vec![object(h(1, 1)), serde_json::json!({"type": "track", "track": hex::encode(track), "from_group": 3})];
|
|
let matched = vec![serde_json::json!({"match": "aa".repeat(32), "candidate": hex::encode(h(0, 2))})];
|
|
// A stranger's takedown, on its own track and on the service's: neither is honoured.
|
|
let (strange, _) = takedown_record(&stranger, exact.clone(), matched.clone());
|
|
let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _st = publish_takedowns(&s, &stranger, 70, std::slice::from_ref(&strange)).await;
|
|
// The service's track carries the stranger's record, then the service's own.
|
|
let (rec, link) = takedown_record(&service, exact.clone(), matched.clone());
|
|
let (m, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _sv = publish_takedowns(&m, &service, 71, &[strange.clone(), rec.clone()]).await;
|
|
let purge_path = d.purge_path(&link);
|
|
assert!(until(STEP, || purge_path.exists()).await, "the purge is written: {:?}", d.counts());
|
|
assert_eq!(d.counts().takedowns, 1, "only the service's takedown is honoured");
|
|
assert_eq!(d.takedown(&strange).map_err(|_| ()), Err(()), "a stranger's record is refused");
|
|
|
|
// FETCH: what it covers is withheld, from the store and the cache alike; the rest
|
|
// verifies against its receipts.
|
|
let all = || Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(0, 0) }));
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
// A group's End of Unknown Range runs through its last withheld object, and covers the
|
|
// objects before it (as a moderation's does): groups 3 and 4 are withheld whole.
|
|
let unknown = vec![Location::new(0, 2), Location::new(1, 1), Location::new(3, OBJECTS - 1), Location::new(4, OBJECTS - 1)];
|
|
let want: std::collections::BTreeSet<(u64, u64)> = [(1, 2), (2, 0), (2, 1), (2, 2)].into_iter().collect();
|
|
let f = fetch_verified(&v, &t, &publisher, all()).await;
|
|
assert_eq!(f.unknown, unknown);
|
|
assert_eq!(f.objects.keys().copied().collect::<std::collections::BTreeSet<_>>(), want);
|
|
// Its bytes are gone from every file of the track.
|
|
let on_disk = |bytes: &[u8]| -> bool {
|
|
let dir = d.track_dir(&media).unwrap();
|
|
std::fs::read_dir(dir).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(bytes.len()).any(|w| w == bytes))
|
|
};
|
|
for (g, o) in [(1, 1), (0, 2), (3, 0), (4, 2)] {
|
|
assert!(!on_disk(&object_payload("td", g, o)), "({g}, {o}) erased from disk");
|
|
}
|
|
assert!(on_disk(&object_payload("td", 2, 0)), "what it does not cover stays");
|
|
// The purge: signed with the relay's key, naming each object and the track.
|
|
let purge = std::fs::read_to_string(&purge_path).unwrap();
|
|
let pv: serde_json::Value = serde_json::from_str(&purge).unwrap();
|
|
let sig: [u8; 64] = hex::decode(pv["sig"].as_str().unwrap()).unwrap().try_into().unwrap();
|
|
let relay_key = server.relay().summary_key().to_bytes();
|
|
assert_eq!(pv["signer_pubkey"], hex::encode(relay_key));
|
|
assert!(joule_moq_relay::takedown::verify_strict(&relay_key, &joule_moq_relay::takedown::signing_bytes(&pv).unwrap(), &sig));
|
|
assert_eq!(pv["takedown"], hex::encode(link));
|
|
assert_eq!(pv["kind"], "wai-takedown-purge");
|
|
let named: Vec<String> = pv["subjects"].as_array().unwrap().iter().map(|x| joule_moq_relay::takedown::jcs(x).unwrap()).collect();
|
|
let mut sorted = named.clone();
|
|
sorted.sort();
|
|
assert_eq!(named, sorted, "ascending");
|
|
assert_eq!(named.len(), 3, "the object, the candidate and the track");
|
|
assert!(named.contains(&joule_moq_relay::takedown::jcs(&object(h(1, 1))).unwrap()) && named.contains(&joule_moq_relay::takedown::jcs(&object(h(0, 2))).unwrap()));
|
|
|
|
// Across a restart: the takedown holds, the purge is the same bytes, and an identical
|
|
// copy published on another track is never stored or served.
|
|
for c in [&p, &s, &m, &v] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
assert_eq!((d.counts().takedowns, d.counts().purges_signed), (1, 0), "{:?}", d.counts());
|
|
assert_eq!(std::fs::read_to_string(&purge_path).unwrap(), purge);
|
|
let (v, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let f = fetch_verified(&v, &t, &publisher, all()).await;
|
|
assert_eq!(f.unknown, unknown);
|
|
assert_eq!(f.objects.keys().copied().collect::<std::collections::BTreeSet<_>>(), want);
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pub2 = publish_tracks(&p, &t2, None).await;
|
|
publish_payloads(&p, &publisher, &t2, 0, vec![object_payload("td", 1, 1), object_payload("copy", 0, 1), object_payload("copy", 0, 2)]).await;
|
|
let media2 = (t2.ns.clone(), t2.media.clone());
|
|
assert!(until(STEP, || d.groups(&media2).contains(&0)).await, "{:?}", d.counts());
|
|
flush(&server).await;
|
|
let f = fetch_verified(&v, &t2, &publisher, all()).await;
|
|
assert_eq!(f.objects.keys().copied().collect::<Vec<_>>(), vec![(0, 1), (0, 2)], "the identical copy is withheld");
|
|
let dir2 = d.track_dir(&media2).unwrap();
|
|
let copy = object_payload("td", 1, 1);
|
|
assert!(!std::fs::read_dir(dir2).unwrap().flatten().any(|e| std::fs::read(e.path()).unwrap().windows(copy.len()).any(|w| w == copy)), "and never stored");
|
|
for c in [&p, &v] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
let _ = std::fs::remove_dir_all(&key_dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// A grant laid out as live-rooms §4.1 lays it out, with these media bits and powers.
|
|
fn grant(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, powers: u8) -> Vec<u8> {
|
|
grant_with(authority, room, principal, publish, powers, &[])
|
|
}
|
|
|
|
/// The same grant with these attributes (live-rooms §4.4), already laid out.
|
|
fn grant_with(authority: &SigningKey, room: [u8; 32], principal: &SigningKey, publish: u8, powers: u8, attrs: &[u8]) -> Vec<u8> {
|
|
let now = now_s();
|
|
let mut g = b"WRG1\x01".to_vec();
|
|
g.extend_from_slice(&room);
|
|
g.extend_from_slice(&principal.verifying_key().to_bytes());
|
|
g.extend_from_slice(&[0; 40]);
|
|
g.extend_from_slice(&(now - 60).to_le_bytes());
|
|
g.extend_from_slice(&(now + 600).to_le_bytes());
|
|
g.extend_from_slice(&[0, publish, 1, 0, powers, 0]);
|
|
g.extend_from_slice(&[0; 32]);
|
|
g.extend_from_slice(b"WEP1\x01");
|
|
g.extend_from_slice(&[0; 32]);
|
|
g.extend_from_slice(&[0; 28]);
|
|
g.extend_from_slice(&[0xff; 16]);
|
|
g.extend_from_slice(&[0; 3]);
|
|
g.extend_from_slice(&[0, 0]); // no profile, no origin
|
|
g.extend_from_slice(&(attrs.len() as u16).to_le_bytes());
|
|
g.extend_from_slice(attrs);
|
|
let mut m = joule_moq::grant_proof::GRANT_DOMAIN.to_vec();
|
|
m.extend_from_slice(&g);
|
|
g.extend_from_slice(&authority.sign(&m).to_bytes());
|
|
g
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn seeking_back_needs_the_seek_power() {
|
|
timeout(Duration::from_secs(120), async {
|
|
const GROUPS: u64 = 6;
|
|
let dir = store_dir("seek");
|
|
let authority = SigningKey::from_bytes(&[0x5a; 32]);
|
|
let (publisher, plain, seeker) = (SigningKey::from_bytes(&[0x5b; 32]), SigningKey::from_bytes(&[0x5c; 32]), SigningKey::from_bytes(&[0x5d; 32]));
|
|
let room = [0x5e; 32];
|
|
let t = room_tracks(room, &publisher);
|
|
let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
|
|
let server = relay(|c| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
|
|
})
|
|
.await;
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
|
|
let _pub = publish_tracks(&p, &t, None).await;
|
|
for g in 0..GROUPS {
|
|
publish_multi(&p, &publisher, &t, g, "s").await;
|
|
}
|
|
written(&server, 2 * GROUPS).await;
|
|
let (a, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &plain, 0, 0), &plain)).await;
|
|
let (b, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &seeker, 0, 1 << 4), &seeker)).await;
|
|
|
|
// Without the power: refused below the live edge, a FETCH and a fill alike.
|
|
let refused = a.fetch(t.ns.clone(), t.media.clone(), range((0, 0), GROUPS - 1)).await.expect_err("below the live edge");
|
|
assert!(refused.contains("seek_not_granted") && refused.contains(&joule_moq::draft22::RequestErrorCode::UNAUTHORIZED.to_string()), "{refused}");
|
|
let fill = Parameters::new().with(Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(1, 0) }));
|
|
let params = Parameters::new().with(Parameter::LocationFilter(LocationFilter::NextObject)).with(Parameter::FillParameters(fill));
|
|
let refused = a.subscribe(t.ns.clone(), t.media.clone(), params).await.expect_err("a fill below the live edge");
|
|
assert!(refused.contains("seek_not_granted"), "{refused}");
|
|
// At the live edge: served.
|
|
let f = fetch_verified(&a, &t, &publisher, range((GROUPS - 1, 0), GROUPS - 1)).await;
|
|
assert_eq!(f.objects.len() as u64, OBJECTS);
|
|
// With it: the window.
|
|
let f = fetch_verified(&b, &t, &publisher, range((0, 0), GROUPS - 1)).await;
|
|
assert_eq!(f.objects.len() as u64, GROUPS * OBJECTS);
|
|
assert!(server.relay().stats().seek_refused >= 2);
|
|
for c in [&p, &a, &b] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
/// The attribute `minor` = `true` (live-rooms §4.4): a short key and a long value.
|
|
fn minor_attr() -> Vec<u8> {
|
|
[&[5u8][..], b"minor", &4u16.to_le_bytes(), b"true"].concat()
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_minors_room_is_never_kept() {
|
|
timeout(Duration::from_secs(120), async {
|
|
let dir = store_dir("minors");
|
|
let authority = SigningKey::from_bytes(&[0x6a; 32]);
|
|
let (publisher, other, minor, watcher) = (SigningKey::from_bytes(&[0x6b; 32]), SigningKey::from_bytes(&[0x6c; 32]), SigningKey::from_bytes(&[0x6d; 32]), SigningKey::from_bytes(&[0x6e; 32]));
|
|
let (room, clear) = ([0x6f; 32], [0x70; 32]);
|
|
let t = room_tracks(room, &publisher);
|
|
let tc = room_tracks(clear, &other);
|
|
let cfg = dvr_config(&dir, 900, 1 << 30, vec![]);
|
|
let config = |c: &mut RelayConfig| {
|
|
c.cache.max_bytes = CACHE;
|
|
c.dvr = Some(cfg.clone());
|
|
c.auth.required = true;
|
|
c.auth.authorities = vec![Authority { key: hexkey(&authority), rooms: vec![] }];
|
|
};
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
|
|
let (o, _) = connect(&server, Via::WebSocket, with(grant(&authority, clear, &other, 0b010, 0), &other)).await;
|
|
let _pt = publish_tracks(&p, &t, None).await;
|
|
let _pc = publish_tracks(&o, &tc, None).await;
|
|
for g in 0..2 {
|
|
publish_multi(&p, &publisher, &t, g, "m").await;
|
|
publish_multi(&o, &other, &tc, g, "c").await;
|
|
}
|
|
written(&server, 8).await;
|
|
let media = (t.ns.clone(), t.media.clone());
|
|
let clear_media = (tc.ns.clone(), tc.media.clone());
|
|
assert_eq!(d.groups(&media), vec![0, 1], "kept until a minor publishes in the room");
|
|
let media_dir = d.track_dir(&media).expect("the room's track, kept so far");
|
|
|
|
// A minor's grant that publishes nothing: nothing changes.
|
|
let (w, _) = connect(&server, Via::WebSocket, with(grant_with(&authority, room, &watcher, 0, 0, &minor_attr()), &watcher)).await;
|
|
publish_multi(&p, &publisher, &t, 2, "m").await;
|
|
assert!(until(STEP, || d.groups(&media).contains(&2)).await, "a minor's grant that publishes nothing keeps the window");
|
|
assert_eq!(d.counts().rooms_sealed, 0);
|
|
|
|
// A minor's publishing grant, proved: the room is forgotten and never kept again.
|
|
let (m, _) = connect(&server, Via::WebSocket, with(grant_with(&authority, room, &minor, 0b011, 0, &minor_attr()), &minor)).await;
|
|
assert!(until(STEP, || d.counts().rooms_sealed == 1).await, "{:?}", d.counts());
|
|
flush(&server).await;
|
|
let receipts = (t.ns.clone(), t.receipts.clone());
|
|
for k in [&media, &receipts] {
|
|
assert!(d.groups(k).is_empty() && d.track_dir(k).is_none(), "{k:?} is kept");
|
|
}
|
|
assert!(!media_dir.exists(), "the room's media is deleted from disk");
|
|
assert_eq!(std::fs::metadata(dir.join("sealed.wdl")).unwrap().len(), 64, "the room is recorded among those never kept");
|
|
publish_multi(&p, &publisher, &t, 3, "m").await;
|
|
publish_multi(&o, &other, &tc, 2, "c").await;
|
|
assert!(until(STEP, || server.relay().largest(&media) == Some(Location::new(3, OBJECTS - 1))).await, "the room is still delivered");
|
|
assert!(until(STEP, || d.groups(&clear_media).contains(&2)).await);
|
|
flush(&server).await;
|
|
assert!(d.groups(&media).is_empty(), "nothing more of the room is written");
|
|
assert_eq!(d.groups(&clear_media), vec![0, 1, 2], "the other room keeps its window");
|
|
|
|
// Across a restart, the room stays forgotten.
|
|
for c in [&p, &o, &w, &m] {
|
|
c.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
let server = relay(config).await;
|
|
let d = server.relay().dvr().unwrap().clone();
|
|
assert_eq!(d.counts().rooms_sealed, 1, "{:?}", d.counts());
|
|
// The restarted relay listens on another port: its proofs name the new target.
|
|
let target = format!("//{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
|
|
let with = |g: Vec<u8>, k: &SigningKey| ClientOptions { grants: vec![(g, k.clone())], target: target.clone(), ..ClientOptions::default() };
|
|
let (p, _) = connect(&server, Via::WebSocket, with(grant(&authority, room, &publisher, 0b010, 0), &publisher)).await;
|
|
let _pt = publish_tracks(&p, &t, None).await;
|
|
publish_multi(&p, &publisher, &t, 4, "m").await;
|
|
assert!(until(STEP, || server.relay().largest(&media) == Some(Location::new(4, OBJECTS - 1))).await);
|
|
flush(&server).await;
|
|
assert!(d.groups(&media).is_empty() && d.track_dir(&media).is_none());
|
|
assert_eq!(d.groups(&clear_media), vec![0, 1, 2]);
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
})
|
|
.await
|
|
.expect("in time");
|
|
}
|
|
|
|
#[test]
|
|
fn a_window_past_a_day_is_refused() {
|
|
let cfg = dvr_config(&PathBuf::from("/nonexistent"), 86_401, 1, vec![]);
|
|
let mut c = RelayConfig::loopback();
|
|
c.dvr = Some(cfg);
|
|
assert!(joule_moq_relay::Relay::new(c).is_err());
|
|
}
|