143 lines
7.5 KiB
Rust
143 lines
7.5 KiB
Rust
//! A room's sealed objects (WAI `extensions/confidentiality.md`) through this relay, which
|
|
//! has no notion of them: three members each publish sealed video with its receipts over
|
|
//! the sealed bytes, the door holds each publisher's namespace to its own key and checks
|
|
//! every object against its receipt, and every member opens what the others sent while a
|
|
//! subscriber outside the group verifies every receipt and opens nothing. The relay
|
|
//! forwards each object's payload and properties byte for byte.
|
|
|
|
mod common;
|
|
|
|
use common::*;
|
|
use ed25519_dalek::SigningKey;
|
|
use joule_moq::draft22::data::{ObjectStatus as WireStatus, SubgroupHeader, SubgroupId, SubgroupObject};
|
|
use joule_moq::draft22::Parameters;
|
|
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder};
|
|
use joule_moq::secure_objects::room::{Incoming, Keyring, NotOpened, Outgoing, Roster};
|
|
use joule_moq::secure_objects::{Plaintext, Refusal, Suite};
|
|
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackName};
|
|
use joule_moq_relay::client::{Client, ClientOptions};
|
|
use joule_moq_relay::config::DoorRule;
|
|
|
|
const GROUPS: u64 = 6;
|
|
const EPOCH: u64 = 3;
|
|
const ROOM: [u8; 32] = [0x5e; 32];
|
|
|
|
/// Publish group `g` of `t`: the receipts over the sealed payload first, then the sealed
|
|
/// object, whose Immutable Properties carry its Key ID and no receipt (confidentiality §4.3).
|
|
async fn publish_sealed(c: &Client, key: &SigningKey, t: &Tracks, g: u64, properties: joule_moq::draft22::kvp::Properties, payload: &[u8]) {
|
|
let hash = ObjectReceipt::hash_payload(payload);
|
|
let mut b = ReceiptBuilder::new(TrackAlias(t.receipt_alias), GroupId(g));
|
|
b.add_labelled_object(hash, 0, "sealed-publisher", "2026-10-19T00:00:00Z", vec![0]).unwrap();
|
|
let receipt = ObjectReceipt {
|
|
track_alias: TrackAlias(t.receipt_alias),
|
|
group_id: GroupId(g),
|
|
object_id: ObjectId(0),
|
|
content_hash: hash,
|
|
joules_micro: 0,
|
|
origin: "sealed-publisher".into(),
|
|
merkle_proof: b.proof_for(0).unwrap(),
|
|
energy_class: Some(vec![0]),
|
|
};
|
|
let group = b.finalize_labelled(key, "sealed-publisher", "2026-10-19T00:00:01Z", None).unwrap();
|
|
let bundle = joule_moq::carriage::ReceiptBundle { group, objects: vec![receipt] };
|
|
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: true, end_of_group: true, first_object: true };
|
|
c.send_subgroup(mh, &[SubgroupObject { object_id: 0, properties, status: WireStatus::Normal, payload: payload.to_vec() }]).await.expect("the object");
|
|
}
|
|
|
|
async fn room_of_three(via: Via) {
|
|
let keys: Vec<SigningKey> = [0xa1u8, 0xb2, 0xc3].iter().map(|b| SigningKey::from_bytes(&[*b; 32])).collect();
|
|
let principals: Vec<[u8; 32]> = keys.iter().map(|k| k.verifying_key().to_bytes()).collect();
|
|
let roster: Roster = principals.iter().enumerate().map(|(i, p)| (i as u32, *p)).collect();
|
|
let secret = [0x77; 32];
|
|
let rings: Vec<Keyring> = principals
|
|
.iter()
|
|
.map(|p| {
|
|
let mut k = Keyring::new(Suite::AesGcm128Sha256, *p);
|
|
k.enter(EPOCH, secret, roster.clone());
|
|
k
|
|
})
|
|
.collect();
|
|
let room = hex::encode(ROOM);
|
|
let server = relay(|c| {
|
|
c.receipts.door = vec![DoorRule { prefix: vec![room.clone(), "*".into(), "*".into()], trust: vec![], trust_field: Some(1), hold_ms: 5_000 }];
|
|
})
|
|
.await;
|
|
let tracks: Vec<Tracks> = principals.iter().enumerate().map(|(i, p)| Tracks::new(ns(&[&room, &hex::encode(p), "video"]), "0", 10 * (i as u64 + 1), 7 + i as u64, 64)).collect();
|
|
|
|
let mut pubs = Vec::new();
|
|
let mut held = Vec::new();
|
|
for t in &tracks {
|
|
let (c, e) = connect(&server, via, ClientOptions::default()).await;
|
|
held.push(publish_tracks(&c, t, None).await);
|
|
pubs.push((c, e));
|
|
}
|
|
// The three members, and one subscriber outside the group.
|
|
let mut subs = Vec::new();
|
|
let mut views = Vec::new();
|
|
for _ in 0..4 {
|
|
let (s, e) = connect(&server, via, ClientOptions::default()).await;
|
|
let mut v = Vec::new();
|
|
for t in &tracks {
|
|
v.push(subscribe_both(&s, t, Parameters::new()).await);
|
|
}
|
|
views.push(v);
|
|
subs.push((s, e));
|
|
}
|
|
|
|
// Every member seals and publishes its groups.
|
|
let mut sent = Vec::new();
|
|
for (i, t) in tracks.iter().enumerate() {
|
|
let name = TrackName::from_str("0");
|
|
for g in 0..GROUPS {
|
|
let pt = Plaintext::with_digest(payload(&format!("member {i}"), g, 700));
|
|
let s = rings[i].seal(&Outgoing { ns: &t.ns, name: &name, group: g, object: 0, priority: t.priority, plaintext: &pt, immutable: &[], mutable: &[] }).unwrap().unwrap();
|
|
publish_sealed(&pubs[i].0, &keys[i], t, g, s.properties.clone(), &s.payload).await;
|
|
sent.push((i, g, s));
|
|
}
|
|
}
|
|
let reads = futures::future::join_all(subs.iter().map(|(s, _)| collect(s, STEP * 2, |got| got.len() as u64 >= 2 * 3 * GROUPS)));
|
|
let got = reads.await;
|
|
|
|
for (j, (g, v)) in got.iter().zip(&views).enumerate() {
|
|
let mut ring = if j < 3 { rings[j].clone() } else { Keyring::new(Suite::AesGcm128Sha256, [0x0f; 32]) };
|
|
for (i, t) in tracks.iter().enumerate() {
|
|
// Every receipt verifies over the sealed bytes the relay forwarded.
|
|
let (objects, bundles, _, _) = verify(g, &v[i], &keys[i]);
|
|
assert_eq!((objects, bundles), (GROUPS as usize, GROUPS as usize), "subscriber {j}, publisher {i}");
|
|
let name = TrackName::from_str("0");
|
|
for x in g.iter().filter(|x| x.header.track_alias == v[i].media_alias) {
|
|
let o = &x.objects[0];
|
|
let (_, _, s) = sent.iter().find(|(k, gg, _)| *k == i && *gg == x.header.group_id).unwrap();
|
|
assert_eq!(o.payload, s.payload, "the sealed payload, byte for byte");
|
|
assert_eq!(o.properties, s.properties, "the properties, the Key ID among them, byte for byte");
|
|
let inc = Incoming { ns: &t.ns, name: &name, group: x.header.group_id, object: 0, priority: x.header.publisher_priority.unwrap(), properties: &o.properties, payload: &o.payload };
|
|
match ring.open(&inc, None) {
|
|
Ok(op) if j < 3 => assert_eq!(op.plaintext.payload, payload(&format!("member {i}"), x.header.group_id, 700)),
|
|
Err(NotOpened::Refused(Refusal::EpochNotJoined)) if j == 3 => {}
|
|
other => panic!("subscriber {j}, publisher {i}: {other:?}"),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert_eq!(server.relay().stats().refused_at_door, 0, "every sealed object passed the door on its receipt");
|
|
drop(held);
|
|
for (s, _) in &subs {
|
|
s.close(0, "done");
|
|
}
|
|
for (p, _) in &pubs {
|
|
p.close(0, "done");
|
|
}
|
|
server.shutdown().await;
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn sealed_objects_over_websocket() {
|
|
tokio::time::timeout(std::time::Duration::from_secs(120), room_of_three(Via::WebSocket)).await.expect("the room finishes in time");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn sealed_objects_over_quic() {
|
|
tokio::time::timeout(std::time::Duration::from_secs(120), room_of_three(Via::Quic)).await.expect("the room finishes in time");
|
|
}
|