open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/messaging_room.rs
Transaction Science 01f3aac11f
Some checks failed
wai-web reach builds / wai-web reach · does this change touch it (pull_request) Successful in 4s
joule-moq-web browser client / joule-moq-web · does this change touch it (pull_request) Successful in 6s
joule-moq-room room service / joule-moq-room · does this change touch it (pull_request) Successful in 8s
wai-web wasm32 / does this change touch it (pull_request) Has been cancelled
wai-web wasm32 / wai-web wasm32 build · corpus, menu and writers (pull_request) Has been cancelled
wai-quantum studio and desktop / does this change touch it (pull_request) Has been cancelled
wai-quantum studio and desktop / studio · browser (pull_request) Has been cancelled
wai-quantum studio and desktop / desktop app · macos (pull_request) Has been cancelled
wai-pack / wai-pack · does this change touch it (pull_request) Has been cancelled
wai-pack / wai-pack · linux (pull_request) Has been cancelled
wai-pack / wai-pack · linux-arm64 (pull_request) Has been cancelled
wai-pack / wai-pack · macos (pull_request) Has been cancelled
wai-pack / wai-pack · independent verifier (Python) (pull_request) Has been cancelled
wai-pack / wai-pack · the floor through a media prober (pull_request) Has been cancelled
wai-pack / wai-pack · dependency licences (pull_request) Has been cancelled
wai-web reach builds / wai-web reach builds · the package of record, every output as recorded (pull_request) Has been cancelled
wai-ops-gpu integer operators on GPUs / does this change touch it (pull_request) Has been cancelled
wai-ops-gpu integer operators on GPUs / builds · no adapter is not a pass (linux) (pull_request) Has been cancelled
wai-ops-gpu integer operators on GPUs / Metal GPU and browser WebGPU (macos) (pull_request) Has been cancelled
wai-web reach builds / wai-web reach builds · this tree's wai-web, every rung on its engine line (pull_request) Has been cancelled
WAI energy receipts / does this change touch it (pull_request) Has been cancelled
WAI energy receipts / energy receipts · linux (pull_request) Has been cancelled
WAI energy receipts / energy receipts · linux-arm64 (pull_request) Has been cancelled
WAI energy receipts / energy receipts · macos (pull_request) Has been cancelled
WAI conformance / does this change touch it (pull_request) Has been cancelled
WAI conformance / cargo-class corpora · linux (pull_request) Has been cancelled
WAI conformance / cargo-class corpora · linux-arm64 (pull_request) Has been cancelled
WAI conformance / cargo-class corpora · macos (pull_request) Has been cancelled
WAI conformance / quantum core and corpus under wasm32-wasip2 (wasmtime) (pull_request) Has been cancelled
WAI conformance / v2 vocabulary · manifests validate as stated (pull_request) Has been cancelled
WAI conformance / independent verifiers · claims, JWP, key discovery and integer payloads (Python) (pull_request) Has been cancelled
WAI conformance / each feature of the independent verifiers' steps compiles alone (pull_request) Has been cancelled
WAI conformance / delivery telemetry · CMCD and CMSD against an independent library (pull_request) Has been cancelled
WAI conformance / cross-arch byte-equality · linux (pull_request) Has been cancelled
WAI conformance / cross-arch byte-equality · linux-arm64 (pull_request) Has been cancelled
WAI conformance / cross-arch byte-equality · macos (pull_request) Has been cancelled
WAI conformance / codec wrappers · fuzz smoke, text decode limits, C entry points, provenance, the semantic path (pull_request) Has been cancelled
WAI conformance / C2PA manifest emission · signed, read back, validated (pull_request) Has been cancelled
standards workspaces build + test / which workspaces this change touches (pull_request) Has been cancelled
standards workspaces build + test / macOS energy backends (pull_request) Has been cancelled
joule-moq-web browser client / joule-moq-web · browser end-to-end (macos) (pull_request) Has been cancelled
joule-moq-room room service / joule-moq-room, every target (pull_request) Has been cancelled
joule-code conformance gate / joule-moq-ingest, every target (pull_request) Has been cancelled
joule-code conformance gate / joule-moq-relay, every target (pull_request) Has been cancelled
repository checks / lockfiles, advisories, tests in CI, workflow hygiene, spec and section references, public text (pull_request) Has been cancelled
joule-code conformance gate / Build + round-trip C1-C4 vectors (pull_request) Has been cancelled
live messaging: regenerate the corpus, a removed member's held-back message
The corpus is regenerated from its seeds since history entries name whether
the principal was admitted. The room test now has a removed member post a
message it sealed before its removal: the service refuses it (not_a_member)
and a verifier refuses it (sender_not_member). The crate README names the
messages and call pages and their e2e scenarios. Clippy notes in the slice's
files are resolved.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-26 11:38:21 -04:00

506 lines
27 KiB
Rust

//! A conversation kept by the room service (WAI live-messaging, `[messaging]`), its members
//! keying through a real MLS group (`joule-moq-mls`) and sealing with `joule-moq`'s secure
//! objects, as pages do:
//!
//! - three members join, key through the service's log and message one another; each opens
//! what the others seal, and `wai-rs` verifies every message as §11 does;
//! - one goes offline; the others go on, the service restarts from its history store, a
//! fourth member joins and a member other than the lowest commits its Add (§7.2); the one
//! offline returns and catches up from the store: every message of both epochs opens;
//! - a thread, reactions, an edit and a retraction, verified, and the retracted message kept
//! in verification, stated retracted;
//! - acknowledgements of delivery and reading taken, and one of another member refused;
//! - a member removed: the next commit excludes it, it opens nothing sealed after its
//! removal, the store serves it nothing past its removal (`not_a_member`) and takes nothing
//! more from it;
//! - every record in the log verifies offline, and the store holds no plaintext.
//!
//! And `wai-rs`'s sealing of a message object (confidentiality §4, as live-messaging §4.2 uses
//! it) and `joule-moq`'s secure objects open what the other seals.
use ed25519_dalek::SigningKey;
use joule_moq::draft22::kvp::Properties;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq::secure_objects::room::{Incoming, Keyring, Outgoing};
use joule_moq::secure_objects::{Plaintext, Suite};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_mls::{CommitRule, Received, RoomMember};
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;
use wai::live_messaging::{self as lm, kind, Conversation, MessageEntry, Open, Outcome};
const STEP: Duration = Duration::from_secs(20);
const STAMP: &str = "2026-10-26T00:00:00Z";
fn key(i: u8) -> SigningKey {
SigningKey::from_bytes(&[i; 32])
}
fn pk(k: &SigningKey) -> [u8; 32] {
k.verifying_key().to_bytes()
}
fn b64(b: &[u8]) -> String {
gp::base64url(b)
}
/// One HTTP/1.1 request: the status, the headers (names in lowercase) and the JSON.
async fn http(addr: std::net::SocketAddr, method: &str, path: &str, headers: &[(&str, String)], body: Option<&Value>) -> (u16, BTreeMap<String, String>, Value) {
let mut s = tokio::net::TcpStream::connect(addr).await.unwrap();
let b = body.map(|v| v.to_string()).unwrap_or_default();
let extra: String = headers.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
let req = format!("{method} {path} HTTP/1.1\r\nHost: room.test\r\nCF-Connecting-IP: 10.0.0.1\r\n{extra}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len());
s.write_all(req.as_bytes()).await.unwrap();
let mut out = Vec::new();
timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time").unwrap();
let text = String::from_utf8(out).unwrap();
let (head, body) = text.split_once("\r\n\r\n").expect("a response");
let status: u16 = head.split(' ').nth(1).unwrap().parse().unwrap();
let hs = head.lines().skip(1).filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect();
(status, hs, serde_json::from_str(body).unwrap_or(Value::Null))
}
/// A principal making requests: its key, its grant (none before it joins), and the challenge
/// the service gave it last (live-rooms §10.5.3).
struct Who {
key: SigningKey,
grant: Option<Vec<u8>>,
challenge: Option<[u8; 32]>,
}
impl Who {
fn new(k: SigningKey) -> Who {
Who { key: k, grant: None, challenge: None }
}
/// A request with this principal's grant (or none: a proof naming no grant) and a proof
/// over the last challenge, asked again once with a fresh challenge when refused 401.
async fn req(&mut self, addr: std::net::SocketAddr, method: &str, path: &str, body: Option<&Value>) -> (u16, Value) {
for _ in 0..2 {
let ch = match self.challenge {
Some(c) => c,
None => {
let (_, _, v) = http(addr, "GET", "/msg/challenge", &[], None).await;
hex::decode(v["challenge"].as_str().unwrap()).unwrap().try_into().unwrap()
}
};
let digest = self.grant.as_ref().map_or([0; 32], |g| gp::grant_digest_and_principal(g).unwrap().0);
let proof = GrantProof::sign(digest, &self.key, ch, &gp::context(method, "room.test", path)).unwrap();
let mut hs = vec![(gp::HEADER_PROOF, b64(&proof.to_bytes()))];
if let Some(g) = &self.grant {
hs.push(("Authorization", format!("Bearer {}", b64(g))));
}
let (st, h, v) = http(addr, method, path, &hs, body).await;
self.challenge = h.get(&gp::HEADER_CHALLENGE.to_ascii_lowercase()).and_then(|c| gp::from_base64url(c)).and_then(|c| c.try_into().ok());
if st != 401 {
return (st, v);
}
}
panic!("{method} {path}: refused twice for its proof");
}
}
/// A member as a page is one: its requests, its MLS state, its place in the log, and the
/// conversation as it verifies it.
struct M {
who: Who,
m: RoomMember,
at: u64,
view: Conversation,
room: [u8; 32],
t0_us: u64,
seen: Vec<Received>,
/// The original payloads it opened, by position.
opened: BTreeMap<u64, Vec<u8>>,
}
fn now_us() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64
}
fn entry_of(v: &Value) -> MessageEntry {
let h = |k: &str| hex::decode(v[k].as_str().unwrap()).unwrap();
MessageEntry { author: h("author").try_into().unwrap(), group: v["group"].as_u64().unwrap(), properties: h("properties"), payload: h("payload"), receipts: h("receipts") }
}
fn ns_of(room: &[u8; 32], author: &[u8; 32]) -> TrackNamespace {
TrackNamespace::from_strs([hex::encode(room).as_str(), hex::encode(author).as_str(), "messages"])
}
impl M {
async fn join(addr: std::net::SocketAddr, k: SigningKey) -> M {
let mut who = Who::new(k.clone());
let (st, v) = who.req(addr, "POST", "/msg/join", Some(&json!({}))).await;
assert_eq!(st, 200, "{v}");
who.grant = Some(hex::decode(v["grant"].as_str().unwrap()).unwrap());
let c = &v["conversation"];
let h = |k: &str| hex::decode(c[k].as_str().unwrap()).unwrap();
let room: [u8; 32] = h("room").try_into().unwrap();
let clock: [u8; 32] = h("clock").try_into().unwrap();
let view = Conversation::new(&h("descriptor"), Some(&h("record")), &h("statement")).expect("the conversation verifies");
let mut m = RoomMember::new(room, clock, k, Suite::AesGcm128Sha256);
m.set_commit_rule(CommitRule::AnyMember);
let mut me = M { who, m, at: 0, view, room, t0_us: c["t0_us"].as_u64().unwrap(), seen: vec![], opened: BTreeMap::new() };
let kp = me.m.key_package().unwrap();
let (st, v) = me.who.req(addr, "POST", "/msg/key-package", Some(&json!({"key_package": hex::encode(kp)}))).await;
assert_eq!(st, 200, "{v}");
me
}
fn me(&self) -> [u8; 32] {
pk(&self.who.key)
}
/// Read the log to its end and act, until nothing is left to read or do. Its status:
/// 200, or the refusal of a read.
async fn pump(&mut self, addr: std::net::SocketAddr) -> u16 {
loop {
let (st, v) = self.who.req(addr, "GET", &format!("/msg/log?from={}&wait_ms=0", self.at), None).await;
if st != 200 {
return st;
}
let entries = v["entries"].as_array().unwrap().clone();
for e in &entries {
let seq = e["seq"].as_u64().unwrap();
assert!(seq >= self.at, "the log goes forward");
if let Some(hx) = e["event"].as_str() {
let rec = hex::decode(hx).unwrap();
let ev = wai::live_events::Event::parse(&rec).unwrap();
ev.verify_signature().expect("the clock's record verifies");
if let Ok(wai::live_events::Payload::Lifecycle { subject, .. }) = wai::live_events::parse_payload(ev.kind, &ev.payload, 1024) {
match wai::live_events::kind_name(ev.kind) {
Some("join") => self.m.event(&joule_moq_mls::Event::Join { subject, keyed: true }),
Some("leave") => self.m.event(&joule_moq_mls::Event::Leave { subject, reason: 1 }),
_ => {}
}
}
} else if let Some(hx) = e["mls"].as_str() {
let r = self.m.receive(&hex::decode(hx).unwrap()).expect("every carried message applies");
self.seen.push(r);
} else if let Some(hx) = e["epoch"].as_str() {
let b = hex::decode(hx).unwrap();
self.view.offer_epoch(seq, &b).expect("the clock's epoch statement verifies");
// §7.3: a member checks the statement against the epoch its group reached.
let s = lm::EpochStatement::read(&b).unwrap();
if let Some(r) = self.m.keyring().roster(s.epoch).filter(|_| self.m.epoch() == Some(s.epoch)) {
let mine: Vec<(u32, [u8; 32])> = r.iter().map(|(l, p)| (*l, *p)).collect();
assert_eq!(mine, s.roster, "the epoch statement names the roster the group reached");
}
} else if let Some(mv) = e.get("message") {
let me = entry_of(mv);
let ns = ns_of(&self.room, &me.author);
let name = TrackName::from_str("0");
let props = Properties::decode(&me.properties, "properties").unwrap();
let open = self.m.open(&Incoming { ns: &ns, name: &name, group: me.group, object: 0, priority: lm::PRIORITY, properties: &props, payload: &me.payload });
match open {
Ok(o) => {
self.opened.insert(seq, o.plaintext.payload.clone());
self.view.offer_message(seq, &me, Open::Opened(&o.plaintext.payload));
}
Err(_) => {
self.view.offer_message(seq, &me, Open::No);
}
}
} else if let Some(hx) = e["ack"].as_str() {
let _ = self.view.offer_ack(seq, &hex::decode(hx).unwrap());
}
self.at = seq + 1;
}
self.at = self.at.max(v["next"].as_u64().unwrap());
let mut acted = false;
if self.m.owes_create() {
let info = self.m.stage_create().unwrap();
if self.post_mls(addr, &[info]).await != 200 {
self.m.withdraw();
}
acted = true;
} else if !self.m.staged() && self.m.is_committer() && self.m.clock_pending() {
let c = self.m.stage_commit(false).unwrap();
let frames: Vec<Vec<u8>> = std::iter::once(c.commit).chain(c.welcome).collect();
if self.post_mls(addr, &frames).await != 200 {
self.m.withdraw();
}
acted = true;
}
if entries.is_empty() && !acted {
return 200;
}
}
}
async fn post_mls(&mut self, addr: std::net::SocketAddr, frames: &[Vec<u8>]) -> u16 {
self.who.req(addr, "POST", "/msg/mls", Some(&json!({"frames": frames.iter().map(hex::encode).collect::<Vec<_>>()}))).await.0
}
/// Seal and post a message of `kind` naming `target`, once the log is read to its end: its
/// status, and its link hash.
async fn send(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> (u16, Value, [u8; 32]) {
let (post, link) = self.prepare(addr, k, target, body).await;
let (st, v) = self.who.req(addr, "POST", "/msg/post", Some(&post)).await;
(st, v, link)
}
/// A message sealed and receipted, as `send` posts it: its post's body and its link hash.
async fn prepare(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> (Value, [u8; 32]) {
self.pump(addr).await;
let (group, parent) = self.view.next_of(&self.me());
let rec = lm::Message::new(&self.who.key, self.room, group, now_us().saturating_sub(self.t0_us), k, target, body.as_bytes());
let ns = ns_of(&self.room, &self.me());
let name = TrackName::from_str("0");
let pt = Plaintext { payload: rec.to_bytes(), ..Default::default() };
let s = self.m.seal(&Outgoing { ns: &ns, name: &name, group, object: 0, priority: lm::PRIORITY, plaintext: &pt, immutable: &[], mutable: &[] }).expect("an epoch to seal under");
let (receipts, link) = lm::receipt(&self.who.key, group, &s.payload, parent, STAMP);
let body = json!({"group": group, "properties": hex::encode(s.properties.to_bytes().unwrap()), "payload": hex::encode(&s.payload), "receipts": hex::encode(receipts)});
(body, link)
}
async fn text(&mut self, addr: std::net::SocketAddr, k: u8, target: [u8; 32], body: &str) -> [u8; 32] {
let (st, v, link) = self.send(addr, k, target, body).await;
assert_eq!(st, 200, "{body}: {v}");
link
}
}
async fn start(cfg: &Config, keys: &Keys) -> (Arc<Service>, std::net::SocketAddr, tokio::task::JoinHandle<()>) {
let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), cfg);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let task = tokio::spawn(http::serve(svc.clone(), listener));
(svc, addr, task)
}
/// Pump every member in turn until the epochs are `want`.
async fn settle(addr: std::net::SocketAddr, ms: &mut [&mut M], want: &[Option<u64>]) {
for _ in 0..100 {
for m in ms.iter_mut() {
m.pump(addr).await;
}
if ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>() == want {
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("epochs {:?}, not {want:?}", ms.iter().map(|m| m.m.epoch()).collect::<Vec<_>>());
}
fn accepted(view: &Conversation, link: &[u8; 32]) -> lm::Accepted {
view.by_link(link).and_then(|v| v.accepted.clone()).expect("an accepted message")
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_conversation_carries_messages_through_a_restart_and_a_removal() {
timeout(Duration::from_secs(240), run()).await.expect("the test finishes in time");
}
async fn run() {
let keys = Keys::generate();
let dir = std::env::temp_dir().join(format!("joule-moq-room-messaging-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://127.0.0.1:1/moq\"\n[messaging]\nstore = \"{}\"\nmax_members = 8\nmax_text = 512\n[rate]\nclient_burst = 64\n", dir.display())).unwrap();
let (svc, addr, task) = start(&cfg, &keys).await;
// Alice, the first, owns the conversation and creates its group; Bob and Carol join, and
// Alice commits their Adds.
let mut a = M::join(addr, key(0x41)).await;
settle(addr, &mut [&mut a], &[Some(0)]).await;
let mut b = M::join(addr, key(0x42)).await;
let mut c = M::join(addr, key(0x43)).await;
settle(addr, &mut [&mut a, &mut b, &mut c], &[Some(1), Some(1), Some(1)]).await;
assert!(a.seen.contains(&Received::Created));
assert!(b.seen.contains(&Received::Joined(1)) && c.seen.contains(&Received::Joined(1)));
// Three members message: each opens what the others seal.
let hello = a.text(addr, kind::TEXT, [0; 32], "hello from alice").await;
b.pump(addr).await;
b.text(addr, kind::TEXT, [0; 32], "hello from bob").await;
c.pump(addr).await;
c.text(addr, kind::TEXT, [0; 32], "hello from carol").await;
for m in [&mut a, &mut b, &mut c] {
m.pump(addr).await;
let texts: Vec<String> = m.view.messages.iter().filter_map(|v| v.accepted.as_ref()).map(|x| String::from_utf8(x.body.clone()).unwrap()).collect();
assert_eq!(texts, ["hello from alice", "hello from bob", "hello from carol"], "every member opens every message");
}
// Carol goes offline. Alice starts a thread, Bob replies, Alice reacts.
let root = a.text(addr, kind::TEXT, [0; 32], "lunch at noon?").await;
b.pump(addr).await;
let reply = b.text(addr, kind::REPLY, root, "yes").await;
a.pump(addr).await;
a.text(addr, kind::REACTION, reply, "+1").await;
let (_, before) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;
// The service restarts from its history store.
task.abort();
drop(svc);
tokio::time::sleep(Duration::from_millis(200)).await;
let (_svc, addr, _task) = start(&cfg, &keys).await;
for m in [&mut a, &mut b, &mut c] {
m.who.challenge = None;
}
let (st, after) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;
assert_eq!(st, 200);
assert_eq!(after["entries"], before["entries"], "the log is as it was, every entry at its position");
// Alice edits her root; Alice goes away, Dave joins and Bob, not the lowest leaf, commits
// his Add (§7.2).
a.pump(addr).await;
a.text(addr, kind::EDIT, root, "lunch at one?").await;
let mut d = M::join(addr, key(0x44)).await;
settle(addr, &mut [&mut b, &mut d], &[Some(2), Some(2)]).await;
assert!(d.seen.contains(&Received::Joined(2)), "dave joined by bob's commit");
let dave_hi = d.text(addr, kind::TEXT, [0; 32], "dave here").await;
b.pump(addr).await;
let oops = b.text(addr, kind::TEXT, [0; 32], "wrong conversation").await;
b.text(addr, kind::RETRACTION, oops, "").await;
a.pump(addr).await;
assert_eq!(a.m.epoch(), Some(2));
a.text(addr, kind::REPLY, root, "see you there").await;
// Carol returns and catches up from the store: both epochs, before and after the restart.
c.pump(addr).await;
assert_eq!(c.m.epoch(), Some(2));
let cv = &c.view;
assert!(cv.messages.iter().all(|v| v.outcome == Outcome::Accepted), "{:?}", cv.messages.iter().map(|v| &v.outcome).collect::<Vec<_>>());
let r = accepted(cv, &root);
assert_eq!((r.text.as_deref(), r.edits, r.replies.len()), (Some(&b"lunch at one?"[..]), 1, 2), "the root edited, with its thread");
assert_eq!(cv.by_link(&root).unwrap().status(), Some("edited"));
let rp = accepted(cv, &reply);
assert_eq!(rp.reactions.iter().map(|(w, body, _)| (*w, body.clone())).collect::<Vec<_>>(), vec![(a.me(), b"+1".to_vec())]);
let x = cv.by_link(&oops).expect("a retracted message stays in verification");
assert_eq!((x.status(), x.accepted.as_ref().unwrap().text.as_ref()), (Some("retracted"), None));
assert!(accepted(cv, &dave_hi).text.is_some());
assert_eq!(c.opened.len(), cv.messages.len(), "carol opened every message, sealed in epochs 1 and 2");
// Acknowledgements: Carol's delivery and reading of Alice's track taken; one naming a
// message it does not hold refused; one of another member's refused.
let (a_next, a_link) = c.view.next_of(&a.me());
let ack = lm::Ack::new(&c.who.key, c.room, a.me(), a_next - 1, a_link.unwrap(), lm::ack::READ, 1).to_bytes();
let (st, v) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&ack)}))).await;
assert_eq!(st, 200, "{v}");
let bad = lm::Ack::new(&c.who.key, c.room, a.me(), a_next - 1, [9; 32], lm::ack::DELIVERED, 2).to_bytes();
let (st, v) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&bad)}))).await;
assert_eq!((st, v["error"].as_str()), (409, Some("ack_unknown")));
let others = lm::Ack::new(&d.who.key, c.room, a.me(), a_next - 1, a_link.unwrap(), lm::ack::DELIVERED, 3).to_bytes();
let (st, _) = c.who.req(addr, "POST", "/msg/ack", Some(&json!({"ack": hex::encode(&others)}))).await;
assert_eq!(st, 403);
a.pump(addr).await;
assert_eq!(a.view.acked.get(&a.me()).and_then(|m| m.get(&c.me())).map(|x| (x.delivered, x.read)), Some((Some(a_next - 1), Some(a_next - 1))));
// Bob seals a message in the epoch he is in, and holds it back.
let (late, _) = b.prepare(addr, kind::TEXT, [0; 32], "sent after my removal").await;
// Alice removes Bob; only an owner may. The next commit excludes him.
let (st, _) = d.who.req(addr, "POST", "/msg/remove", Some(&json!({"principal": hex::encode(b.me())}))).await;
assert_eq!(st, 403, "dave is not an owner");
let (st, v) = a.who.req(addr, "POST", "/msg/remove", Some(&json!({"principal": hex::encode(b.me())}))).await;
assert_eq!(st, 200, "{v}");
settle(addr, &mut [&mut a, &mut c, &mut d], &[Some(3), Some(3), Some(3)]).await;
assert_eq!(b.pump(addr).await, 403, "bob reads what he is owed up to his removal, then nothing: not_a_member");
assert!(b.seen.contains(&Received::Removed(3)), "bob applies the commit that removes him");
let removal_end = b.at;
// Alice writes in the epoch Bob is not in.
let after_removal = a.text(addr, kind::TEXT, [0; 32], "bob cannot read this").await;
c.pump(addr).await;
assert!(accepted(&c.view, &after_removal).text.is_some());
// The store serves Bob nothing past his removal: from where he stopped, nothing more, and
// from the end, not_a_member.
let (st, v) = b.who.req(addr, "GET", &format!("/msg/log?from={removal_end}&wait_ms=0"), None).await;
assert_eq!(st, 403, "{v}");
assert_eq!(v["error"].as_str(), Some("not_a_member"));
// He opens nothing sealed after it, though he holds the object.
let pos = *c.opened.keys().last().unwrap();
let (_, log) = c.who.req(addr, "GET", &format!("/msg/log?from={pos}&wait_ms=0"), None).await;
let e = entry_of(&log["entries"][0]["message"]);
let props = Properties::decode(&e.properties, "properties").unwrap();
let ns = ns_of(&b.room, &e.author);
let opened = b.m.open(&Incoming { ns: &ns, name: &TrackName::from_str("0"), group: e.group, object: 0, priority: lm::PRIORITY, properties: &props, payload: &e.payload });
assert!(opened.is_err(), "a removed member opens nothing of a later epoch: {opened:?}");
// The store takes nothing more from him, not even what he sealed before his removal: he is not
// admitted, and the message is a removed member's (§11.3 step 2).
let (st, v) = b.who.req(addr, "POST", "/msg/post", Some(&late)).await;
assert_eq!((st, v["error"].as_str()), (403, Some("not_a_member")));
let late_entry = MessageEntry {
author: b.me(),
group: late["group"].as_u64().unwrap(),
properties: hex::decode(late["properties"].as_str().unwrap()).unwrap(),
payload: hex::decode(late["payload"].as_str().unwrap()).unwrap(),
receipts: hex::decode(late["receipts"].as_str().unwrap()).unwrap(),
};
c.pump(addr).await;
assert_eq!(c.view.check_post(&late_entry).map(|_| ()), Err(lm::Reason::SenderNotMember), "a verifier refuses a removed member's message");
// He is not admitted again.
let mut again = Who::new(key(0x42));
let (st, v) = again.req(addr, "POST", "/msg/join", Some(&json!({}))).await;
assert_eq!((st, v["error"].as_str()), (403, Some("removed")));
// Every record of the log verifies offline: the whole log as Alice is owed it, every
// message opened with what Alice opened.
a.pump(addr).await;
let (_, full) = a.who.req(addr, "GET", "/msg/log?from=0&wait_ms=0", None).await;
assert_eq!(full["withheld"].as_array().unwrap().len(), 0, "alice, a member throughout, is owed every entry");
let (_, _, conv) = http(addr, "GET", "/msg", &[], None).await;
let h = |k: &str| hex::decode(conv[k].as_str().unwrap()).unwrap();
let mut offline = Conversation::new(&h("descriptor"), Some(&h("record")), &h("statement")).unwrap();
let (mut events, mut messages, mut acks) = (0, 0, 0);
for e in full["entries"].as_array().unwrap() {
let seq = e["seq"].as_u64().unwrap();
if let Some(x) = e["event"].as_str() {
wai::live_events::Event::parse(&hex::decode(x).unwrap()).unwrap().verify_signature().unwrap();
events += 1;
} else if let Some(x) = e["epoch"].as_str() {
offline.offer_epoch(seq, &hex::decode(x).unwrap()).unwrap();
} else if let Some(mv) = e.get("message") {
let p = a.opened.get(&seq).expect("alice opened it");
assert_eq!(offline.offer_message(seq, &entry_of(mv), Open::Opened(p)), Outcome::Accepted);
messages += 1;
} else if let Some(x) = e["ack"].as_str() {
offline.offer_ack(seq, &hex::decode(x).unwrap()).unwrap();
acks += 1;
}
}
assert_eq!((events, messages, acks), (5, 12, 1), "four joins and a leave, every message, one acknowledgement");
assert_eq!(offline.epochs().len(), 4);
// The store holds ciphertext and signed records only.
let journal = std::fs::read(dir.join("journal")).unwrap();
for plain in ["hello from alice", "lunch at", "wrong conversation", "bob cannot read this"] {
assert!(!journal.windows(plain.len()).any(|w| w == plain.as_bytes()), "the store holds no plaintext: {plain}");
}
let _ = hello;
let _ = std::fs::remove_dir_all(&dir);
}
/// `wai-rs`'s sealing of a message object and `joule-moq`'s secure objects agree: each opens
/// what the other seals, under one epoch's secret, at one KID.
#[test]
fn wai_rs_and_secure_objects_seal_alike() {
let room = [7u8; 32];
let alice = key(0x51);
let secret = [0x5c; 32];
let roster: joule_moq::secure_objects::room::Roster = [(0u32, pk(&key(0x50))), (1, pk(&alice))].into();
let mut ring = Keyring::new(Suite::AesGcm128Sha256, pk(&alice));
ring.enter(4, secret, roster.clone());
let record = lm::Message::new(&alice, room, 3, 9, kind::TEXT, [0; 32], b"sealed by wai-rs").to_bytes();
let kid = lm::seal::kid(4, 1);
let (props, sealed) = lm::seal::seal(lm::SUITE_AES128, &secret, &room, &pk(&alice), 3, kid, &record, None);
let ns = ns_of(&room, &pk(&alice));
let name = TrackName::from_str("0");
let p = Properties::decode(&props, "properties").unwrap();
let mut reader = Keyring::new(Suite::AesGcm128Sha256, pk(&key(0x50)));
reader.enter(4, secret, roster);
let o = reader.open(&Incoming { ns: &ns, name: &name, group: 3, object: 0, priority: lm::PRIORITY, properties: &p, payload: &sealed }, None).expect("secure objects open what wai-rs sealed");
assert_eq!(o.plaintext.payload, record);
// And the other way.
let pt = Plaintext { payload: record.clone(), padding: Some(24), ..Default::default() };
let s = ring.seal(&Outgoing { ns: &ns, name: &name, group: 5, object: 0, priority: lm::PRIORITY, plaintext: &pt, immutable: &[], mutable: &[] }).unwrap().unwrap();
assert_eq!(s.kid, kid);
let got = lm::seal::open(lm::SUITE_AES128, &secret, &room, &pk(&alice), 5, &s.properties.to_bytes().unwrap(), &s.payload).expect("wai-rs opens what secure objects sealed");
assert_eq!(got, record);
assert!(lm::seal::open(lm::SUITE_AES128, &[0; 32], &room, &pk(&alice), 5, &s.properties.to_bytes().unwrap(), &s.payload).is_err(), "another secret opens nothing");
}