f8c8613308
`files` on the control surface, beside `sessions`, reachable from a shell with
wai-relay --control: {"op":"files","dir":"<absolute path>","room":"<room>"?} writes the
store's window as .moq and .dat files (Dvr::write_files, draft-jennings-moq-file-05), every
track the store holds or one room's, and answers dir, tracks and files.
- dir is an absolute path that does not exist or is an empty directory, so a request
overwrites nothing. Refused before anything is written, each with its reason:
dir_not_empty, dir_not_absolute, dir_unreadable, no_store, no_tracks.
- Written off the workers that carry sessions.
- tests/dvr_files.rs an_operator_asks_for_a_room_as_files: two rooms recorded, one asked for
with the program and verified from its files alone against the publisher's receipts, then
every track; each refusal planted, an earlier take left byte for byte as it was.
Co-Authored-By: Claude Fable 5.2 <[email protected]>
291 lines
16 KiB
Rust
291 lines
16 KiB
Rust
//! A recorded room as MoQ files (`draft-jennings-moq-file-05`; WAI live-recording §8.7), end
|
|
//! to end: a publisher sends a room's video and its receipts through the relay, the DVR store
|
|
//! keeps them, the store writes its window as `.moq` and `.dat` files, and a reader that holds
|
|
//! only those files and the publisher's key verifies every object against its group's signed
|
|
//! receipt. What the files state is what a FETCH of the same window brings, byte for byte.
|
|
//!
|
|
//! Planted, each refused by the reader with its own reason and nothing of the track
|
|
//! delivered: a data file cut short, a `.moq` file cut short, and a `.moq` file naming a data
|
|
//! file outside its directory; and a payload changed in the data file, which reads (the files
|
|
//! carry no check of their own) and is caught by the receipt.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_moq::carriage::ReceiptBundle;
|
|
use joule_moq::draft22::data::{FetchEntry, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::Parameters;
|
|
use joule_moq::file::{self, FileError, FileObject, FileRefusal};
|
|
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder, ReceiptVerifier};
|
|
use joule_moq::{GroupId, ObjectId, TrackAlias};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::{DvrConfig, DvrWindow};
|
|
use std::path::Path;
|
|
use std::time::Duration;
|
|
use tokio::time::timeout;
|
|
|
|
const GROUPS: u64 = 6;
|
|
const OBJECTS: u64 = 3;
|
|
|
|
fn payload_of(g: u64, o: u64) -> Vec<u8> {
|
|
let mut v = format!("rec:{g}:{o}:").into_bytes();
|
|
v.resize(700 + (g * 13 + o) as usize, (g * 7 + o) as u8);
|
|
v
|
|
}
|
|
|
|
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-11-10T00: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-11-10T00:00:01Z", None).unwrap();
|
|
ReceiptBundle { group, objects }
|
|
}
|
|
|
|
async fn publish(c: &Client, key: &SigningKey, t: &Tracks, g: u64) {
|
|
let payloads: Vec<Vec<u8>> = (0..OBJECTS).map(|o| payload_of(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");
|
|
}
|
|
|
|
/// Verify a directory's media track against its receipts track, from the files alone: the
|
|
/// objects verified, or why not.
|
|
fn verify_files(dir: &Path, t: &Tracks, publisher: &SigningKey) -> Result<usize, String> {
|
|
let objects = file::read_dir(dir).map_err(|e| e.to_string())?;
|
|
let verifier = ReceiptVerifier::with_anchors(vec![publisher.verifying_key()]);
|
|
let of = |name: &joule_moq::TrackName| -> Vec<&FileObject> { objects.iter().filter(|o| o.namespace == t.ns && &o.name == name).collect() };
|
|
let mut verified = 0;
|
|
for m in of(&t.media) {
|
|
let r = of(&t.receipts).into_iter().find(|r| r.group_id == m.group_id).ok_or(format!("no receipts for group {}", m.group_id))?;
|
|
let b = ReceiptBundle::from_payload(&r.payload).map_err(|e| e.to_string())?;
|
|
let vg = verifier.verify_group(&b.group).map_err(|e| e.to_string())?;
|
|
let receipt = b.objects.get(m.object_id as usize).ok_or("no receipt at the object's position")?;
|
|
verifier.verify_payload(&m.payload, receipt, &vg).map_err(|e| format!("group {} object {}: {e}", m.group_id, m.object_id))?;
|
|
verified += 1;
|
|
}
|
|
Ok(verified)
|
|
}
|
|
|
|
fn copy_dir(from: &Path, name: &str) -> std::path::PathBuf {
|
|
let to = store_dir(name);
|
|
std::fs::create_dir_all(&to).unwrap();
|
|
for e in std::fs::read_dir(from).unwrap() {
|
|
let e = e.unwrap();
|
|
std::fs::copy(e.path(), to.join(e.file_name())).unwrap();
|
|
}
|
|
to
|
|
}
|
|
|
|
fn one_with(dir: &Path, ext: &str, holding: &[u8]) -> std::path::PathBuf {
|
|
std::fs::read_dir(dir)
|
|
.unwrap()
|
|
.map(|e| e.unwrap().path())
|
|
.filter(|p| p.extension().is_some_and(|x| x == ext))
|
|
.find(|p| {
|
|
let b = std::fs::read(p).unwrap();
|
|
holding.is_empty() || b.windows(holding.len()).any(|w| w == holding)
|
|
})
|
|
.expect("a file of the take")
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn a_recorded_room_is_written_as_moq_files_and_verifies_from_them() {
|
|
timeout(Duration::from_secs(180), run()).await.expect("the test finishes in time");
|
|
}
|
|
|
|
async fn run() {
|
|
let store = store_dir("files-store");
|
|
let room = [0x61; 32];
|
|
let publisher = SigningKey::from_bytes(&[0x62; 32]);
|
|
let t = Tracks::new(ns(&[&hex::encode(room), &hex::encode(publisher.verifying_key().to_bytes()), "video"]), "0", 10, 5, 128);
|
|
let cfg = DvrConfig { path: store.clone(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 900, max_bytes: 1 << 24 }], room_keys: vec![], takedown_keys: vec![] };
|
|
let server = relay(|c| 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(&p, &publisher, &t, g).await;
|
|
}
|
|
let d = server.relay().dvr().expect("a DVR store").clone();
|
|
assert!(until(STEP, || d.counts().groups_written >= 2 * GROUPS).await, "every group stored: {:?}", d.counts());
|
|
|
|
// The recorder writes the room as files.
|
|
let take = store_dir("files-take");
|
|
let (d2, take2) = (d.clone(), take.clone());
|
|
let written = tokio::task::spawn_blocking(move || d2.write_files(&[], &take2)).await.unwrap().expect("the files");
|
|
assert_eq!(written.len(), 2, "a .moq file for the track and one for its receipts track");
|
|
|
|
// Read back from the files alone: every object, verified against its group's receipt.
|
|
assert_eq!(verify_files(&take, &t, &publisher), Ok((GROUPS * OBJECTS) as usize));
|
|
let read = file::read_dir(&take).expect("the take reads");
|
|
let media: Vec<&FileObject> = read.iter().filter(|o| o.name == t.media).collect();
|
|
assert_eq!(media.len() as u64, GROUPS * OBJECTS);
|
|
for o in &media {
|
|
assert_eq!(o.payload, payload_of(o.group_id, o.object_id), "the payload as published");
|
|
assert_eq!(o.forwarding, file::Forwarding::Subgroup(0));
|
|
assert!(o.receive_time_ms.is_some(), "when the relay received its group");
|
|
}
|
|
|
|
// And what the files state is what a FETCH of the window brings.
|
|
let (c, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
for (name, per_group) in [(&t.media, OBJECTS), (&t.receipts, 1)] {
|
|
let (_ok, entries) = c.fetch(t.ns.clone(), name.clone(), Parameters::new()).await.expect("a FETCH of the window");
|
|
let fetched: Vec<(u64, u64, Vec<u8>)> = entries.into_iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some((o.group_id, o.object_id, o.payload)) } else { None }).collect();
|
|
let filed: Vec<(u64, u64, Vec<u8>)> = read.iter().filter(|o| &o.name == name).map(|o| (o.group_id, o.object_id, o.payload.clone())).collect();
|
|
assert_eq!(fetched.len() as u64, GROUPS * per_group);
|
|
assert_eq!(fetched, filed, "the files and the FETCH agree, byte for byte");
|
|
}
|
|
|
|
// ── Planted faults ──
|
|
let marker = payload_of(2, 1);
|
|
// A data file cut short.
|
|
let cut = copy_dir(&take, "files-cut-dat");
|
|
let dat = one_with(&cut, "dat", &marker[..12]);
|
|
let bytes = std::fs::read(&dat).unwrap();
|
|
std::fs::write(&dat, &bytes[..bytes.len() - 1]).unwrap();
|
|
assert!(matches!(file::read_dir(&cut), Err(FileError::Refused(FileRefusal::DataTruncated))), "{:?}", file::read_dir(&cut).map(|o| o.len()));
|
|
// A .moq file cut short.
|
|
let cut = copy_dir(&take, "files-cut-moq");
|
|
let moq = one_with(&cut, "moq", b"");
|
|
let text = std::fs::read(&moq).unwrap();
|
|
std::fs::write(&moq, &text[..text.len() - 7]).unwrap();
|
|
assert!(matches!(file::read_dir(&cut), Err(FileError::Refused(FileRefusal::MetaMalformed))), "{:?}", file::read_dir(&cut).map(|o| o.len()));
|
|
// A .moq file naming a data file outside its directory.
|
|
let out = copy_dir(&take, "files-outside");
|
|
let moq = one_with(&out, "moq", b"");
|
|
let dat_name = one_with(&out, "dat", b"").file_name().unwrap().to_string_lossy().into_owned();
|
|
let other_dat = std::fs::read_dir(&out).unwrap().map(|e| e.unwrap().file_name().to_string_lossy().into_owned()).filter(|n| n.ends_with(".dat")).collect::<Vec<_>>();
|
|
let text = std::fs::read_to_string(&moq).unwrap();
|
|
let named = other_dat.iter().find(|n| text.contains(n.as_str())).unwrap_or(&dat_name);
|
|
std::fs::write(&moq, text.replace(named.as_str(), &format!("../{named}"))).unwrap();
|
|
assert!(matches!(file::read_dir(&out), Err(FileError::Refused(FileRefusal::DataPathRefused))), "{:?}", file::read_dir(&out).map(|o| o.len()));
|
|
// A payload changed in the data file: it reads, and the receipt refuses it.
|
|
let changed = copy_dir(&take, "files-changed");
|
|
let dat = one_with(&changed, "dat", &marker[..12]);
|
|
let mut bytes = std::fs::read(&dat).unwrap();
|
|
let at = bytes.windows(12).position(|w| w == &marker[..12]).unwrap();
|
|
bytes[at + 20] ^= 1;
|
|
std::fs::write(&dat, &bytes).unwrap();
|
|
assert!(file::read_dir(&changed).is_ok(), "the files carry no check of their own");
|
|
let refused = verify_files(&changed, &t, &publisher).unwrap_err();
|
|
assert!(refused.starts_with("group 2 object 1"), "{refused}");
|
|
|
|
c.close(0, "done");
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
}
|
|
|
|
/// One request sent with the program, as an operator sends it from a shell: its exit status
|
|
/// and its answer.
|
|
#[cfg(unix)]
|
|
async fn operator(socket: &Path, request: serde_json::Value) -> (i32, serde_json::Value) {
|
|
let (socket, request) = (socket.to_path_buf(), request.to_string());
|
|
let run = move || std::process::Command::new(env!("CARGO_BIN_EXE_wai-relay")).arg("--control").arg(&socket).arg(&request).output().expect("wai-relay runs");
|
|
let o = timeout(STEP * 2, tokio::task::spawn_blocking(run)).await.expect("in time").expect("it ran");
|
|
(o.status.code().expect("it exited"), serde_json::from_str(String::from_utf8_lossy(&o.stdout).trim()).unwrap_or(serde_json::Value::Null))
|
|
}
|
|
|
|
/// The operator's request (`files` on the control surface, sent with `wai-relay --control`):
|
|
/// two rooms recorded, one room asked for, written into a directory the relay makes, and read
|
|
/// back from the files alone. Then every track, into another. Planted, each refused with its
|
|
/// reason and nothing written: a directory that holds files, a relative path, a room the
|
|
/// store holds nothing of, and a relay that keeps no store.
|
|
#[cfg(unix)]
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn an_operator_asks_for_a_room_as_files() {
|
|
timeout(Duration::from_secs(180), async {
|
|
let store = store_dir("files-op-store");
|
|
let socket = std::env::temp_dir().join(format!("wai-relay-files-{}.sock", std::process::id()));
|
|
let publisher = SigningKey::from_bytes(&[0x72; 32]);
|
|
let pk = hex::encode(publisher.verifying_key().to_bytes());
|
|
let (room_a, room_b) = (hex::encode([0x71; 32]), hex::encode([0x73; 32]));
|
|
let ta = Tracks::new(ns(&[&room_a, &pk, "video"]), "0", 10, 5, 128);
|
|
let tb = Tracks::new(ns(&[&room_b, &pk, "video"]), "0", 20, 15, 128);
|
|
let cfg = DvrConfig { path: store.clone(), window: vec![DvrWindow { prefix: vec!["*".into()], max_age_s: 900, max_bytes: 1 << 24 }], room_keys: vec![], takedown_keys: vec![] };
|
|
let server = relay(|c| {
|
|
c.dvr = Some(cfg.clone());
|
|
c.control.socket = Some(socket.clone());
|
|
})
|
|
.await;
|
|
let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
|
|
let _pubs = (publish_tracks(&p, &ta, None).await, publish_tracks(&p, &tb, None).await);
|
|
for g in 0..GROUPS {
|
|
publish(&p, &publisher, &ta, g).await;
|
|
publish(&p, &publisher, &tb, g).await;
|
|
}
|
|
let d = server.relay().dvr().expect("a DVR store").clone();
|
|
assert!(until(STEP, || d.counts().groups_written >= 4 * GROUPS).await, "every group stored: {:?}", d.counts());
|
|
|
|
// One room, into a directory that does not exist yet.
|
|
let take = store_dir("files-op-take").join("room-a");
|
|
let (code, a) = operator(&socket, serde_json::json!({"op": "files", "dir": take, "room": room_a})).await;
|
|
assert_eq!((code, &a["ok"], &a["tracks"], a["files"].as_array().map(Vec::len)), (0, &serde_json::json!(true), &serde_json::json!(2), Some(2)), "{a}");
|
|
assert_eq!(verify_files(&take, &ta, &publisher), Ok((GROUPS * OBJECTS) as usize));
|
|
let read = file::read_dir(&take).expect("the take reads");
|
|
assert!(read.iter().all(|o| o.namespace == ta.ns), "nothing of the other room");
|
|
for f in a["files"].as_array().unwrap() {
|
|
let f = Path::new(f.as_str().expect("a path"));
|
|
assert!(f.starts_with(&take) && f.is_file() && f.with_extension("dat").is_file(), "{}", f.display());
|
|
}
|
|
// Every track the store holds.
|
|
let all = store_dir("files-op-all").join("all");
|
|
let (code, a) = operator(&socket, serde_json::json!({"op": "files", "dir": all})).await;
|
|
assert_eq!((code, &a["tracks"], a["files"].as_array().map(Vec::len)), (0, &serde_json::json!(4), Some(4)), "{a}");
|
|
assert_eq!((verify_files(&all, &ta, &publisher), verify_files(&all, &tb, &publisher)), (Ok((GROUPS * OBJECTS) as usize), Ok((GROUPS * OBJECTS) as usize)));
|
|
|
|
// ── Planted ──
|
|
let listing = |dir: &Path| {
|
|
let mut v: Vec<(std::ffi::OsString, Vec<u8>)> = std::fs::read_dir(dir).unwrap().map(|e| e.unwrap()).map(|e| (e.file_name(), std::fs::read(e.path()).unwrap())).collect();
|
|
v.sort();
|
|
v
|
|
};
|
|
let before = listing(&take);
|
|
let refused = async |request: serde_json::Value, why: &str| {
|
|
let (code, a) = operator(&socket, request.clone()).await;
|
|
assert_eq!((code, &a["ok"]), (1, &serde_json::json!(false)), "{request}: {a}");
|
|
assert!(a["error"].as_str().is_some_and(|e| e.starts_with(why)), "{request}: {a}");
|
|
};
|
|
// A directory that holds files: nothing of it is overwritten.
|
|
refused(serde_json::json!({"op": "files", "dir": take, "room": room_b}), "dir_not_empty").await;
|
|
assert_eq!(listing(&take), before, "the earlier take is as it was");
|
|
// A relative path.
|
|
refused(serde_json::json!({"op": "files", "dir": "take-here"}), "dir_not_absolute").await;
|
|
assert!(!Path::new("take-here").exists());
|
|
// A room the store holds nothing of: no directory is made for it.
|
|
let none = store_dir("files-op-none").join("none");
|
|
refused(serde_json::json!({"op": "files", "dir": none, "room": hex::encode([0x7f; 32])}), "no_tracks").await;
|
|
assert!(!none.exists(), "nothing was written");
|
|
// A room that is not one.
|
|
refused(serde_json::json!({"op": "files", "dir": none, "room": "zz"}), "").await;
|
|
p.close(0, "done");
|
|
server.shutdown().await;
|
|
|
|
// A relay that keeps no store.
|
|
let socket2 = std::env::temp_dir().join(format!("wai-relay-files-none-{}.sock", std::process::id()));
|
|
let bare = relay(|c| c.control.socket = Some(socket2.clone())).await;
|
|
let (code, a) = operator(&socket2, serde_json::json!({"op": "files", "dir": none})).await;
|
|
assert_eq!(code, 1, "{a}");
|
|
assert!(a["error"].as_str().is_some_and(|e| e.starts_with("no_store")), "{a}");
|
|
bare.shutdown().await;
|
|
})
|
|
.await
|
|
.expect("the test finishes in time");
|
|
}
|