pub mod persist;
pub mod rendezvous;
pub mod sas;
use anyhow::Context;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard};
const INVITE_SCHEME: &str = "mcpmesh-invite:";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Invite {
pub secret: [u8; 32],
pub inviter_id: [u8; 32],
pub inviter_addr_json: String,
pub nickname: String,
pub services: Vec<String>,
pub expires_at_epoch: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub app_label: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub peer_nickname: Option<String>,
#[serde(default = "mcpmesh_local_api::one_use")]
pub uses_remaining: u32,
}
pub const MAX_APP_LABEL_LEN: usize = 256;
impl Invite {
pub fn encode(&self) -> String {
let wire = Self {
peer_nickname: None,
..self.clone()
};
let json = serde_json::to_vec(&wire).expect("invite serializes");
format!(
"{INVITE_SCHEME}{}",
data_encoding::BASE32_NOPAD.encode(&json)
)
}
pub fn decode(line: &str) -> anyhow::Result<Self> {
let payload = line.strip_prefix(INVITE_SCHEME).ok_or_else(|| {
anyhow::anyhow!("not an mcpmesh invite (missing {INVITE_SCHEME} scheme)")
})?;
let json = data_encoding::BASE32_NOPAD
.decode(payload.as_bytes())
.context("invite payload is not valid base32")?;
serde_json::from_slice(&json).context("invite payload is not a valid invite")
}
}
#[derive(Debug)]
pub enum Redeem {
Ok(Invite),
Expired,
Unknown,
Unavailable,
}
#[derive(Default)]
pub struct LiveInvites {
inner: Mutex<HashMap<[u8; 32], Invite>>,
file: Option<persist::InviteFile>,
writes: tokio::sync::Mutex<()>,
}
impl LiveInvites {
pub fn new() -> Self {
Self::default()
}
pub fn load(path: impl Into<std::path::PathBuf>, now_epoch: u64) -> Self {
let file = persist::InviteFile::new(path);
let on_disk = file.load(0).len();
let live = file.load(now_epoch);
let reaped = live.len() != on_disk;
let map: HashMap<[u8; 32], Invite> = live.into_iter().map(|i| (i.secret, i)).collect();
if reaped && let Err(e) = file.store(&map.values().collect::<Vec<_>>()) {
tracing::warn!(%e, "could not rewrite the invite file after reaping expired invites");
}
Self {
inner: Mutex::new(map),
file: Some(file),
writes: tokio::sync::Mutex::new(()),
}
}
async fn persist(&self, snapshot: Vec<Invite>) -> std::io::Result<()> {
let Some(file) = self.file.clone() else {
return Ok(());
};
crate::util::blocking("join invite persist", move || {
file.store(&snapshot.iter().collect::<Vec<_>>())
})
.await
.map_err(std::io::Error::other)?
}
fn guard(&self) -> MutexGuard<'_, HashMap<[u8; 32], Invite>> {
self.inner.lock().expect("LiveInvites mutex poisoned")
}
pub async fn mint(&self, invite: Invite) -> std::io::Result<()> {
let _w = self.writes.lock().await;
let secret = invite.secret;
let (displaced, snapshot) = {
let mut map = self.guard();
let displaced = map.insert(secret, invite);
(displaced, map.values().cloned().collect::<Vec<_>>())
};
match self.persist(snapshot).await {
Ok(()) => Ok(()),
Err(e) => {
let mut map = self.guard();
match displaced {
Some(prev) => map.insert(secret, prev),
None => map.remove(&secret),
};
Err(e)
}
}
}
pub fn peek_live(&self, secret: &[u8; 32], now_epoch: u64) -> bool {
self.peek_live_alias(secret, now_epoch).is_some()
}
pub fn peek_live_alias(&self, secret: &[u8; 32], now_epoch: u64) -> Option<Option<String>> {
self.guard()
.get(secret)
.filter(|inv| inv.expires_at_epoch >= now_epoch)
.map(|inv| inv.peer_nickname.clone())
}
pub async fn try_redeem(&self, secret: &[u8; 32], now_epoch: u64) -> Redeem {
let _w = self.writes.lock().await;
let (outcome, snapshot) = {
let mut map = self.guard();
match map.get(secret) {
None => return Redeem::Unknown,
Some(inv) if inv.expires_at_epoch < now_epoch => {
map.remove(secret);
(Redeem::Expired, map.values().cloned().collect::<Vec<_>>())
}
Some(_) => {
let entry = map.get_mut(secret).expect("present under lock");
let Some(left) = entry.uses_remaining.checked_sub(1) else {
return Redeem::Unknown;
};
entry.uses_remaining = left;
let inv = if left == 0 {
map.remove(secret).expect("present under lock")
} else {
entry.clone()
};
(Redeem::Ok(inv), map.values().cloned().collect::<Vec<_>>())
}
}
};
if let Err(e) = self.persist(snapshot).await {
tracing::warn!(%e, "could not record an invite redemption; refusing it rather than \
risking a restart re-issuing the use");
let mut map = self.guard();
if let Redeem::Ok(inv) = &outcome {
let mut restored = inv.clone();
restored.uses_remaining = restored.uses_remaining.saturating_add(1);
map.insert(*secret, restored);
}
return Redeem::Unavailable;
}
outcome
}
pub fn count(&self) -> usize {
self.guard().len()
}
pub async fn remove_expired(&self, now_epoch: u64) {
let _w = self.writes.lock().await;
let (changed, snapshot) = {
let mut map = self.guard();
let before = map.len();
map.retain(|_, inv| inv.expires_at_epoch >= now_epoch);
(
map.len() != before,
map.values().cloned().collect::<Vec<_>>(),
)
};
if changed {
inv_persist_burn(self, snapshot).await;
}
}
}
async fn inv_persist_burn(reg: &LiveInvites, snapshot: Vec<Invite>) {
if let Err(e) = reg.persist(snapshot).await {
tracing::warn!(
%e,
"could not persist an invite burn; it may reappear after a restart until it expires"
);
}
}
#[cfg(test)]
mod tests {
#[tokio::test]
async fn an_invite_survives_a_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let inv = sample_invite(1, 9_000);
let first = LiveInvites::load(&path, 1_000);
first.mint(inv.clone()).await.unwrap();
drop(first);
let after = LiveInvites::load(&path, 2_000);
assert_eq!(after.count(), 1, "the invite must still be outstanding");
assert!(
matches!(after.try_redeem(&inv.secret, 2_000).await, Redeem::Ok(_)),
"and must still be redeemable — an invite that survives but cannot be spent is no \
better than one that did not survive"
);
}
#[tokio::test]
async fn a_redemption_is_not_undone_by_a_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let inv = sample_invite(2, 9_000);
let first = LiveInvites::load(&path, 1_000);
first.mint(inv.clone()).await.unwrap();
assert!(matches!(
first.try_redeem(&inv.secret, 1_000).await,
Redeem::Ok(_)
));
drop(first);
let after = LiveInvites::load(&path, 1_000);
assert_eq!(after.count(), 0);
assert!(
matches!(after.try_redeem(&inv.secret, 1_000).await, Redeem::Unknown),
"a spent single-use credential must not be resurrected by a restart"
);
}
#[tokio::test]
async fn an_expired_invite_is_dropped_on_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let dead = sample_invite(3, 5_000);
let live = sample_invite(4, 9_000);
let first = LiveInvites::load(&path, 1_000);
first.mint(dead.clone()).await.unwrap();
first.mint(live.clone()).await.unwrap();
drop(first);
let after = LiveInvites::load(&path, 6_000); assert_eq!(after.count(), 1, "only the live one survives");
assert!(matches!(
after.try_redeem(&dead.secret, 6_000).await,
Redeem::Unknown
));
assert!(matches!(
after.try_redeem(&live.secret, 6_000).await,
Redeem::Ok(_)
));
}
#[tokio::test]
async fn a_mint_that_cannot_persist_fails_and_rolls_back() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
std::fs::create_dir(&path).unwrap();
let reg = LiveInvites::load(&path, 1_000);
let inv = sample_invite(5, 9_000);
assert!(
reg.mint(inv.clone()).await.is_err(),
"a mint that cannot promise the TTL it advertises must not report success"
);
assert_eq!(reg.count(), 0, "and must leave no phantom behind");
assert!(matches!(
reg.try_redeem(&inv.secret, 1_000).await,
Redeem::Unknown
));
}
#[tokio::test]
async fn an_all_expired_file_is_rewritten_empty() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let first = LiveInvites::load(&path, 1_000);
first.mint(sample_invite(7, 5_000)).await.unwrap();
first.mint(sample_invite(8, 5_000)).await.unwrap();
drop(first);
let after = LiveInvites::load(&path, 9_000); assert_eq!(after.count(), 0);
let on_disk = std::fs::read_to_string(&path).unwrap();
assert_eq!(
on_disk, "[]",
"expired bearer secrets must not sit at rest — with nothing live, no later mutation \
will ever rewrite this file: {on_disk}"
);
}
#[tokio::test]
async fn reaping_expired_invites_is_persisted() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let reg = LiveInvites::load(&path, 1_000);
reg.mint(sample_invite(9, 5_000)).await.unwrap();
reg.mint(sample_invite(10, 9_000)).await.unwrap();
reg.remove_expired(6_000).await;
assert_eq!(reg.count(), 1, "memory drops the expired one");
let on_disk = persist::InviteFile::new(&path).load(0);
assert_eq!(
on_disk.len(),
1,
"and so must DISK — otherwise the reaped secret is still readable: {on_disk:?}"
);
assert_eq!(on_disk[0].expires_at_epoch, 9_000);
}
#[tokio::test]
async fn a_multi_use_invite_admits_exactly_its_quota() {
let reg = LiveInvites::new();
let mut inv = sample_invite(11, 9_000);
inv.uses_remaining = 3;
reg.mint(inv.clone()).await.unwrap();
for expected_left in [2, 1, 0] {
match reg.try_redeem(&inv.secret, 1_000).await {
Redeem::Ok(got) => assert_eq!(
got.uses_remaining, expected_left,
"each redemption decrements, and the caller sees what is left"
),
other => panic!("redemption within quota must succeed, got {other:?}"),
}
}
assert_eq!(
reg.count(),
0,
"at zero the invite is BURNED, exactly as single-use"
);
assert!(
matches!(reg.try_redeem(&inv.secret, 1_000).await, Redeem::Unknown),
"an exhausted invite answers Unknown — the same answer a secret that never existed \
gets, so exhausting one is not an oracle"
);
}
#[tokio::test]
async fn a_single_use_invite_still_burns_on_first_redemption() {
let reg = LiveInvites::new();
let inv = sample_invite(12, 9_000);
assert_eq!(inv.uses_remaining, 1, "the sample IS the default shape");
reg.mint(inv.clone()).await.unwrap();
assert!(matches!(
reg.try_redeem(&inv.secret, 1_000).await,
Redeem::Ok(_)
));
assert_eq!(reg.count(), 0);
}
#[tokio::test]
async fn the_remaining_use_count_survives_a_restart() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("invites.json");
let mut inv = sample_invite(13, 9_000);
inv.uses_remaining = 3;
let first = LiveInvites::load(&path, 1_000);
first.mint(inv.clone()).await.unwrap();
assert!(matches!(
first.try_redeem(&inv.secret, 1_000).await,
Redeem::Ok(_)
));
drop(first);
let after = LiveInvites::load(&path, 1_000);
match after.try_redeem(&inv.secret, 1_000).await {
Redeem::Ok(got) => assert_eq!(
got.uses_remaining, 1,
"the restart must not restore spent uses — 3 minted, 1 spent, 2 left, so this \
redemption leaves 1"
),
other => panic!("expected a redemption, got {other:?}"),
}
}
#[test]
fn an_invite_line_without_a_use_count_decodes_as_single_use() {
let mut v = serde_json::to_value(sample_invite(14, 9_000)).unwrap();
v.as_object_mut().unwrap().remove("uses_remaining");
let old: Invite = serde_json::from_value(v).expect("an older invite must still decode");
assert_eq!(
old.uses_remaining, 1,
"an invite minted before #87 is single-use, not unusable"
);
}
#[tokio::test]
async fn expiry_beats_remaining_uses() {
let reg = LiveInvites::new();
let mut inv = sample_invite(15, 5_000);
inv.uses_remaining = 10;
reg.mint(inv.clone()).await.unwrap();
assert!(
matches!(reg.try_redeem(&inv.secret, 6_000).await, Redeem::Expired),
"uses left does not outlive the TTL"
);
assert_eq!(reg.count(), 0, "and it is removed");
}
#[tokio::test]
async fn a_ram_only_registry_is_unchanged() {
let reg = LiveInvites::new();
let inv = sample_invite(6, 9_000);
reg.mint(inv.clone())
.await
.expect("a RAM-only mint cannot fail");
assert_eq!(reg.count(), 1);
assert!(matches!(
reg.try_redeem(&inv.secret, 1_000).await,
Redeem::Ok(_)
));
}
use super::*;
fn sample_invite(secret: u8, expires_at_epoch: u64) -> Invite {
Invite {
secret: [secret; 32],
inviter_id: [3u8; 32],
inviter_addr_json: "{\"id\":\"abc\",\"addrs\":[]}".into(),
nickname: "alice".into(),
services: vec!["notes".into()],
expires_at_epoch,
app_label: None,
uses_remaining: 1,
peer_nickname: None,
}
}
#[test]
fn invite_roundtrips_through_the_line_encoding() {
let inv = sample_invite(7, 1_800_000_000);
let line = inv.encode(); assert!(line.starts_with("mcpmesh-invite:"));
let back = Invite::decode(&line).unwrap();
assert_eq!(back, inv);
assert!(Invite::decode("mcpmesh-invite:!!!not-valid").is_err());
assert!(Invite::decode("notaninvite").is_err());
}
#[test]
fn invite_carries_an_opaque_app_label_additively() {
let mut inv = sample_invite(9, 1_800_000_000);
inv.app_label = Some("urn:kb-mesh:node:abc123".into());
let back = Invite::decode(&inv.encode()).unwrap();
assert_eq!(back.app_label.as_deref(), Some("urn:kb-mesh:node:abc123"));
assert_eq!(back, inv);
let no_label = sample_invite(9, 1_800_000_000);
assert!(no_label.app_label.is_none());
let json = serde_json::to_vec(&no_label).unwrap();
assert!(!String::from_utf8_lossy(&json).contains("app_label"));
let line = format!(
"mcpmesh-invite:{}",
data_encoding::BASE32_NOPAD.encode(&json)
);
assert_eq!(Invite::decode(&line).unwrap().app_label, None);
}
#[test]
fn decode_rejects_hostile_payloads_without_panicking() {
assert!(Invite::decode("mcpmesh-invite:").is_err());
let not_invite = data_encoding::BASE32_NOPAD.encode(b"{\"nope\":1}");
assert!(Invite::decode(&format!("mcpmesh-invite:{not_invite}")).is_err());
}
#[tokio::test]
async fn mint_then_redeem_valid_burns_the_invite() {
let live = LiveInvites::default();
let inv = sample_invite(7, 1_800_000_000);
let secret = inv.secret;
live.mint(inv.clone()).await.unwrap();
assert_eq!(live.count(), 1);
match live.try_redeem(&secret, 1_000_000_000).await {
Redeem::Ok(got) => {
assert_eq!(got.uses_remaining, 0, "spent");
assert_eq!(
Invite {
uses_remaining: inv.uses_remaining,
..got
},
inv
);
}
other => panic!("expected Ok, got {other:?}"),
}
assert!(matches!(
live.try_redeem(&secret, 1_000_000_000).await,
Redeem::Unknown
));
assert_eq!(live.count(), 0);
}
#[tokio::test]
async fn redeem_unknown_secret_is_unknown_and_leaves_other_invites_untouched() {
let live = LiveInvites::default();
let inv = sample_invite(7, 1_800_000_000);
live.mint(inv).await.unwrap();
assert!(matches!(
live.try_redeem(&[9u8; 32], 1_000_000_000).await,
Redeem::Unknown
));
assert_eq!(
live.count(),
1,
"unknown secret must not burn a live invite"
);
}
#[tokio::test]
async fn redeem_expired_secret_is_expired_and_removed() {
let live = LiveInvites::default();
let inv = sample_invite(7, 1_000);
let secret = inv.secret;
live.mint(inv).await.unwrap();
assert!(matches!(
live.try_redeem(&secret, 2_000).await,
Redeem::Expired
));
assert_eq!(live.count(), 0);
}
#[tokio::test]
async fn remove_expired_drops_only_the_stale_invites() {
let live = LiveInvites::default();
live.mint(sample_invite(1, 1_000)).await.unwrap(); live.mint(sample_invite(2, 9_000)).await.unwrap(); assert_eq!(live.count(), 2);
live.remove_expired(2_000).await;
assert_eq!(live.count(), 1);
assert!(matches!(
live.try_redeem(&[2u8; 32], 2_000).await,
Redeem::Ok(_)
));
}
}