1use std::path::{Path, PathBuf};
33
34use serde::{Deserialize, Serialize};
35
36use crate::identity::{self, Identity};
37use crate::to_hex;
38
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
40pub struct Member {
41 pub key: String,
42 pub role: String,
43}
44
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46pub struct Membership {
47 pub collective: String,
48 pub admin: String,
50 pub version: u64,
51 #[serde(default, skip_serializing_if = "String::is_empty")]
53 pub prev: String,
54 pub members: Vec<Member>,
55 #[serde(default, skip_serializing_if = "Vec::is_empty")]
57 pub revoked: Vec<Revoked>,
58 #[serde(default, skip_serializing_if = "Vec::is_empty")]
60 pub merged: Vec<String>,
61 #[serde(default)]
62 pub sig: String,
63 #[serde(default, skip_serializing_if = "Vec::is_empty")]
66 pub also: Vec<Also>,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70pub struct Revoked {
71 pub key: String,
72 pub at_ms: i64,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
78pub struct Also {
79 pub from: String,
80 pub key: String,
81 pub at_ms: i64,
82}
83
84fn body(m: &Membership) -> Vec<u8> {
86 let unsigned = Membership { sig: String::new(), also: Vec::new(), ..m.clone() };
87 let mut b = b"borg-membership-v1\n".to_vec();
88 b.extend(serde_json::to_vec(&unsigned).expect("membership serialises"));
89 b
90}
91
92pub fn path(home: &Path) -> PathBuf {
93 identity::dir(home).join("membership.json")
94}
95
96pub fn history_path(home: &Path) -> PathBuf {
99 identity::dir(home).join("membership-history.jsonl")
100}
101
102impl Membership {
103 pub fn found(collective: &str, admin: &Identity) -> Membership {
105 let mut m = Membership {
106 collective: collective.into(),
107 admin: admin.public(),
108 version: 1,
109 prev: String::new(),
110 members: vec![Member { key: admin.public(), role: "admin".into() }],
111 revoked: Vec::new(),
112 merged: Vec::new(),
113 sig: String::new(),
114 also: Vec::new(),
115 };
116 m.sig = admin.sign(&body(&m));
117 m
118 }
119
120 pub fn id(&self) -> String {
122 let mut b = body(self);
123 b.extend(self.sig.as_bytes());
124 to_hex(blake3::hash(&b).as_bytes())
125 }
126
127 pub fn signed(&self) -> Membership {
129 Membership { also: Vec::new(), ..self.clone() }
130 }
131
132 pub fn verify(&self) -> Result<(), String> {
134 identity::verify(&self.admin, &body(self), &self.sig).map_err(|e| format!("membership of {}: {e}", self.collective))?;
135 if self.listed(&self.admin) != Some("admin") {
136 return Err(format!("membership of {}: its admin is not listed as admin", self.collective));
137 }
138 Ok(())
139 }
140
141 fn listed(&self, key: &str) -> Option<&str> {
143 self.members.iter().find(|m| m.key == key).map(|m| m.role.as_str())
144 }
145
146 pub fn role_of(&self, key: &str) -> Option<&str> {
148 if self.also.iter().any(|a| a.key == key) {
149 return None;
150 }
151 self.listed(key)
152 }
153
154 pub fn invite(&mut self, admin: &Identity, key: &str, role: &str) -> Result<(), String> {
157 self.check_admin(admin, "invite")?;
158 identity::check_key(key)?;
159 let parent = self.id();
160 match self.members.iter_mut().find(|m| m.key == key) {
161 Some(m) => m.role = role.into(),
162 None => self.members.push(Member { key: key.into(), role: role.into() }),
163 }
164 self.revoked.retain(|r| r.key != key);
165 self.also.retain(|a| a.key != key);
166 self.resign(admin, parent)
167 }
168
169 pub fn revoke(&mut self, admin: &Identity, key: &str, now_ms: i64) -> Result<(), String> {
171 self.check_admin(admin, "revoke")?;
172 identity::check_key(key)?;
173 if key == admin.public() {
174 return Err("an admin cannot revoke itself; another admin can".into());
175 }
176 if self.role_of(key).is_none() {
177 return Err(format!("{key} is not a member of {}", self.collective));
178 }
179 let parent = self.id();
180 self.members.retain(|m| m.key != key);
181 self.revoked.retain(|r| r.key != key);
182 self.revoked.push(Revoked { key: key.into(), at_ms: now_ms });
183 self.resign(admin, parent)
184 }
185
186 fn check_admin(&self, admin: &Identity, what: &str) -> Result<(), String> {
187 self.verify()?;
188 if self.role_of(&admin.public()) != Some("admin") {
189 return Err(format!("only an admin of {} can {what}", self.collective));
190 }
191 Ok(())
192 }
193
194 fn resign(&mut self, admin: &Identity, parent: String) -> Result<(), String> {
196 self.version = self.version.checked_add(1).ok_or("membership version exhausted")?;
197 for a in std::mem::take(&mut self.also) {
198 if self.members.iter().any(|m| m.key == a.key) {
199 self.members.retain(|m| m.key != a.key);
200 self.revoked.retain(|r| r.key != a.key);
201 self.revoked.push(Revoked { key: a.key.clone(), at_ms: a.at_ms });
202 } else if let Some(r) = self.revoked.iter_mut().find(|r| r.key == a.key) {
203 r.at_ms = r.at_ms.min(a.at_ms);
204 }
205 if !self.merged.contains(&a.from) {
206 self.merged.push(a.from);
207 }
208 }
209 self.prev = parent;
210 self.admin = admin.public();
211 self.sig = admin.sign(&body(self));
212 self.verify()
213 }
214
215 pub fn superseded_by(&self, newer: &Membership) -> bool {
217 newer.verify().is_ok()
218 && newer.collective == self.collective
219 && Some(newer.version) == self.version.checked_add(1)
220 && newer.prev == self.id()
221 && self.role_of(&newer.admin) == Some("admin")
222 }
223
224 pub fn resolve(&self, docs: &[Membership]) -> (Membership, Vec<Membership>) {
227 let mut pool: Vec<(String, Membership)> =
228 docs.iter().map(Membership::signed).filter(|d| d.collective == self.collective && d.verify().is_ok()).map(|d| (d.id(), d)).collect();
229 pool.sort_by(|a, b| a.0.cmp(&b.0));
230 pool.dedup_by(|a, b| a.0 == b.0);
231 let kids = |p: &Membership| pool.iter().filter(|(_, d)| p.superseded_by(d)).cloned().collect::<Vec<_>>();
232 let mut cur = self.clone();
233 let mut attached = Vec::new();
234 loop {
235 let children = kids(&cur);
236 let Some((_, winner)) = children.first().cloned() else { break };
237 let mut also = cur.also.clone();
238 for (from, loser) in &children[1..] {
239 let mut todo = vec![(cur.signed(), loser.clone())];
241 while let Some((parent, d)) = todo.pop() {
242 for r in d.revoked.iter().filter(|r| !parent.revoked.iter().any(|p| p.key == r.key)) {
243 also.push(Also { from: from.clone(), key: r.key.clone(), at_ms: r.at_ms });
244 }
245 todo.extend(kids(&d).into_iter().map(|(_, c)| (d.clone(), c)));
246 attached.push(d);
247 }
248 }
249 also.retain(|a| !winner.merged.contains(&a.from) && winner.listed(&a.key).is_some());
250 also.sort_by(|a, b| (&a.key, a.at_ms, &a.from).cmp(&(&b.key, b.at_ms, &b.from)));
251 also.dedup();
252 attached.push(winner.clone());
253 cur = Membership { also, ..winner };
254 }
255 (cur, attached)
256 }
257
258 pub fn newest_via(&self, docs: &[Membership]) -> Membership {
260 self.resolve(docs).0
261 }
262
263 pub fn admits(&self, key: &str, ms: i64) -> bool {
266 let cut = self.also.iter().filter(|a| a.key == key).map(|a| a.at_ms).min();
267 (self.listed(key).is_some() && cut.is_none_or(|at| ms < at)) || self.revoked.iter().any(|r| r.key == key && ms < r.at_ms)
268 }
269
270 pub fn history(home: &Path) -> Vec<Membership> {
272 let mut v = recorded(home);
273 if let Ok(Some(cur)) = Membership::load(home) {
274 let cur = cur.signed();
275 if !v.contains(&cur) {
276 v.push(cur);
277 }
278 }
279 v.sort_by_key(|m| m.version);
280 v.dedup();
281 v
282 }
283
284 pub fn adopt(home: &Path, more: &[Membership]) -> Result<(Membership, bool), String> {
288 let held = Membership::load(home)?.ok_or_else(|| format!("no membership in {}", path(home).display()))?;
289 let own = Membership::history(home);
290 let root = own.first().cloned().unwrap_or_else(|| held.signed());
291 let mut pool = own;
292 pool.extend(more.iter().cloned());
293 let (tip, attached) = root.resolve(&pool);
294 if tip == held {
295 return Ok((held, false));
296 }
297 for d in &attached {
298 d.record(home)?;
299 }
300 tip.save(home)?;
301 Ok((tip, true))
302 }
303
304 pub fn load(home: &Path) -> Result<Option<Membership>, String> {
306 let p = path(home);
307 let text = match std::fs::read_to_string(&p) {
308 Ok(t) => t,
309 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
310 Err(e) => return Err(format!("{}: {e}", p.display())),
311 };
312 let m: Membership = serde_json::from_str(&text).map_err(|e| format!("{}: {e}", p.display()))?;
313 m.verify()?;
314 Ok(Some(m))
315 }
316
317 pub fn save(&self, home: &Path) -> Result<(), String> {
318 self.verify()?;
319 let p = path(home);
320 std::fs::create_dir_all(identity::dir(home)).map_err(|e| e.to_string())?;
321 let tmp = p.with_extension("json.tmp");
322 std::fs::write(&tmp, serde_json::to_string_pretty(self).unwrap()).map_err(|e| format!("{}: {e}", tmp.display()))?;
323 std::fs::rename(&tmp, &p).map_err(|e| format!("{}: {e}", p.display()))?;
324 let _ = borg::presence::retain_mismatches(home, |k| self.role_of(k).is_some());
326 let _ = borg::presence::retain_reads(home, |k| self.role_of(k).is_some());
327 let _ = borg::protocol::retain_versions(home, |k| self.role_of(k).is_some());
328 self.record(home)
329 }
330
331 pub fn record(&self, home: &Path) -> Result<(), String> {
333 let signed = self.signed();
334 signed.verify()?;
335 if recorded(home).contains(&signed) {
336 return Ok(());
337 }
338 use std::io::Write;
339 std::fs::create_dir_all(identity::dir(home)).map_err(|e| e.to_string())?;
340 let h = history_path(home);
341 let mut f = std::fs::OpenOptions::new().create(true).append(true).open(&h).map_err(|e| format!("{}: {e}", h.display()))?;
342 writeln!(f, "{}", serde_json::to_string(&signed).unwrap()).map_err(|e| format!("{}: {e}", h.display()))
343 }
344}
345
346fn recorded(home: &Path) -> Vec<Membership> {
348 std::fs::read_to_string(history_path(home))
349 .unwrap_or_default()
350 .lines()
351 .filter_map(|l| serde_json::from_str::<Membership>(l).ok())
352 .map(|m| m.signed())
353 .filter(|m| m.verify().is_ok())
354 .collect()
355}
356
357pub fn parse_docs(text: &str) -> Result<Vec<Membership>, String> {
360 if let Ok(m) = serde_json::from_str::<Membership>(text) {
361 return Ok(vec![m.signed()]);
362 }
363 text.lines().filter(|l| !l.trim().is_empty()).map(|l| serde_json::from_str::<Membership>(l).map(|m| m.signed()).map_err(|e| e.to_string())).collect()
364}
365
366#[cfg(test)]
367mod tests {
368 use super::*;
369
370 fn node(tag: &str) -> (PathBuf, Identity) {
371 let home = std::env::temp_dir().join(format!("borg-mesh-mem-{tag}-{}", std::process::id()));
372 let _ = std::fs::remove_dir_all(&home);
373 let id = Identity::load_or_create(&home).unwrap();
374 (home, id)
375 }
376
377 #[test]
378 fn a_member_that_leaves_takes_its_protocol_records_along() {
379 let ((ha, admin), (hb, bob), (hc, eve)) = (node("pa"), node("pb"), node("pc"));
380 let mut m = Membership::found("work", &admin);
381 m.invite(&admin, &bob.public(), "member").unwrap();
382 m.invite(&admin, &eve.public(), "member").unwrap();
383 m.save(&ha).unwrap();
384 let old = |ms| borg::presence::Mismatch { theirs: "borg/mesh/5".into(), ours: "borg/mesh/10".into(), ms };
385 borg::presence::note_mismatch(&ha, &bob.public(), old(1)).unwrap();
386 borg::presence::note_mismatch(&ha, &eve.public(), old(2)).unwrap();
387 borg::presence::note_reads(&ha, &bob.public(), 1).unwrap();
388 borg::presence::note_reads(&ha, &eve.public(), 2).unwrap();
389 for (k, ms) in [(bob.public(), 1), (eve.public(), 2)] {
390 let v = borg::protocol::PeerVersion { borg: Some("0.7.0".into()), protocol: 15, ms, engineer: None };
391 borg::protocol::note_version(&ha, &k, &v).unwrap();
392 }
393 m.revoke(&admin, &bob.public(), 3).unwrap();
394 m.save(&ha).unwrap();
395 let left: Vec<String> = borg::presence::mismatches(&ha).into_keys().collect();
396 assert_eq!(left, vec![eve.public()], "only the member still listed keeps its warning");
397 let reads: Vec<String> = borg::presence::reads(&ha).into_keys().collect();
398 assert_eq!(reads, vec![eve.public()], "and the line format its node reads");
399 let versions: Vec<String> = borg::protocol::versions(&ha).into_keys().collect();
400 assert_eq!(versions, vec![eve.public()], "and the version it last told");
401 for h in [ha, hb, hc] {
402 let _ = std::fs::remove_dir_all(h);
403 }
404 }
405
406 #[test]
407 fn admin_invites_and_nobody_else_can() {
408 let ((ha, admin), (hb, bob), (hc, eve)) = (node("a"), node("b"), node("c"));
409 let mut m = Membership::found("work", &admin);
410 m.verify().unwrap();
411 assert_eq!(m.role_of(&bob.public()), None);
412 m.invite(&admin, &bob.public(), "member").unwrap();
413 assert_eq!((m.version, m.role_of(&bob.public())), (2, Some("member")));
414 assert!(m.clone().invite(&bob, &eve.public(), "member").is_err(), "a member cannot invite");
415 assert!(m.clone().invite(&admin, "not-a-key", "member").is_err());
416
417 let mut forged = m.clone();
418 forged.members.push(Member { key: eve.public(), role: "member".into() });
419 assert!(forged.verify().is_err(), "an edited document no longer verifies");
420 let mut hijack = Membership::found("work", &eve);
421 hijack.version = 99;
422 assert!(!m.superseded_by(&hijack), "another admin's document is never adopted");
423 let old = Membership::found("work", &admin);
424 assert!(!m.superseded_by(&old) && old.superseded_by(&m), "only newer versions are adopted");
425
426 m.save(&ha).unwrap();
427 assert_eq!(Membership::load(&ha).unwrap(), Some(m));
428 for h in [ha, hb, hc] {
429 let _ = std::fs::remove_dir_all(h);
430 }
431 }
432
433 #[test]
434 fn revocation_and_admin_hand_over() {
435 let ((ha, admin), (hb, bob), (hc, carol)) = (node("ra"), node("rb"), node("rc"));
436 let v1 = Membership::found("work", &admin);
437 let mut m = v1.clone();
438 m.invite(&admin, &bob.public(), "member").unwrap();
439 m.invite(&admin, &carol.public(), "member").unwrap();
440 let v3 = m.clone();
441 assert!(m.clone().revoke(&bob, &carol.public(), 10).is_err(), "a member cannot revoke");
442 assert!(m.clone().revoke(&admin, &admin.public(), 10).is_err(), "an admin cannot revoke itself");
443 m.revoke(&admin, &carol.public(), 1000).unwrap();
444 assert_eq!((m.version, m.role_of(&carol.public())), (4, None));
445 assert!(v3.superseded_by(&m), "the revocation is adopted by holders of the previous version");
446 assert!(m.admits(&carol.public(), 999) && !m.admits(&carol.public(), 1000), "before the revocation only");
447 assert!(m.admits(&bob.public(), 5) && !m.admits("ff", 5));
448
449 let v4 = m.clone();
450 m.invite(&admin, &bob.public(), "admin").unwrap();
453 let v5 = m.clone();
454 m.invite(&bob, &carol.public(), "member").unwrap();
455 assert_eq!((m.version, m.admin.clone()), (6, bob.public()));
456 assert!(!v3.superseded_by(&m), "bob was not an admin in v3");
457 assert_eq!(v3.newest_via(&[m.clone(), v1.clone(), v5.clone()]), v3, "no step skips a version");
458 assert_eq!(v3.newest_via(&[m.clone(), v1.clone(), v5.clone(), v4]), m);
459 assert!(m.admits(&carol.public(), 5000), "re-invited: the revocation is lifted");
460
461 m.save(&ha).unwrap();
462 v1.save(&hb).unwrap();
463 v5.save(&hb).unwrap();
464 assert_eq!(Membership::history(&hb), vec![v1, v5]);
465 for h in [ha, hb, hc] {
466 let _ = std::fs::remove_dir_all(h);
467 }
468 }
469
470 fn sign_as(admin: &Identity, mut d: Membership) -> Membership {
472 d.admin = admin.public();
473 d.sig = admin.sign(&body(&d));
474 d
475 }
476
477 #[test]
478 fn a_version_can_only_move_by_one_and_never_overflow() {
479 let ((ha, a1), (hb, a2)) = (node("max-a"), node("max-b"));
480 let mut v4 = Membership::found("work", &a1);
481 v4.invite(&a1, &a2.public(), "admin").unwrap();
482 let parent = v4.id();
484 let mut maxed = v4.clone();
485 maxed.members.retain(|m| m.key != a1.public());
486 maxed.revoked.push(Revoked { key: a1.public(), at_ms: 1 });
487 maxed.prev = parent;
488 maxed.version = u64::MAX;
489 let maxed = sign_as(&a2, maxed);
490 maxed.verify().unwrap();
491 assert!(!v4.superseded_by(&maxed), "a version jump is never adopted");
492 assert_eq!(v4.newest_via(std::slice::from_ref(&maxed)), v4);
493 let mut wrong_parent = v4.clone();
494 wrong_parent.invite(&a2, &a2.public(), "admin").unwrap();
495 wrong_parent.prev = "00".repeat(32);
496 assert!(!v4.superseded_by(&sign_as(&a2, wrong_parent)), "a child must name its parent");
497 let mut top = maxed.clone();
499 top.members.push(Member { key: a1.public(), role: "admin".into() });
500 let mut top = sign_as(&a1, top);
501 assert!(top.invite(&a1, &a2.public(), "member").unwrap_err().contains("exhausted"));
502 for h in [ha, hb] {
503 let _ = std::fs::remove_dir_all(h);
504 }
505 }
506
507 #[test]
511 fn concurrent_admins_resolve_the_same_way_and_no_revocation_is_lost() {
512 let ((ha, a1), (hb, a2), (hc, c), (hd, d)) = (node("fork-a"), node("fork-b"), node("fork-c"), node("fork-d"));
513 let mut v4 = Membership::found("work", &a1);
514 v4.invite(&a1, &a2.public(), "admin").unwrap();
515 v4.invite(&a1, &c.public(), "member").unwrap();
516 v4.invite(&a1, &d.public(), "admin").unwrap();
517 let mut x = v4.clone();
518 x.revoke(&a1, &c.public(), 100).unwrap();
519 let mut y = v4.clone();
520 y.revoke(&a2, &a1.public(), 200).unwrap();
521 let y5 = y.clone();
522 y.invite(&a2, &d.public(), "member").unwrap();
523 let docs = vec![x.clone(), y5.clone(), y.clone()];
524 let mut rev = docs.clone();
525 rev.reverse();
526 let (tip, attached) = v4.resolve(&docs);
527 assert_eq!(tip, v4.resolve(&rev).0, "order does not matter");
528 assert_eq!(attached.len(), 3);
529 assert_eq!(tip.role_of(&c.public()), None, "A1's revocation of C stands");
530 assert_eq!(tip.role_of(&a1.public()), None, "A2's revocation of A1 stands");
531 assert!(tip.admits(&c.public(), 99) && !tip.admits(&c.public(), 100));
532 assert_eq!(x.newest_via(&docs).version, x.version, "a loser alone has nowhere to go");
534 assert_eq!(v4.resolve(&[x.clone()]).0.resolve(&docs).0.role_of(&c.public()), None);
535
536 let survivor = [&a1, &a2, &d].into_iter().find(|k| tip.role_of(&k.public()) == Some("admin")).expect("an admin survives");
538 let mut next = tip.clone();
539 next.invite(survivor, &hd_key(&hd), "member").unwrap();
540 assert!(next.also.is_empty() && !next.merged.is_empty());
541 assert_eq!(next.role_of(&c.public()), None);
542 let mut all = docs.clone();
543 all.push(next.clone());
544 assert_eq!(v4.resolve(&all).0, next, "folded revocations are not applied twice");
545 for h in [ha, hb, hc, hd] {
546 let _ = std::fs::remove_dir_all(h);
547 }
548 }
549
550 fn hd_key(h: &Path) -> String {
551 Identity::load_or_create(&h.join("e")).unwrap().public()
552 }
553}