use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use mcpmesh::allowlist::{AllowlistGate, PeerEntry, PeerStore};
use mcpmesh::config::Config;
use mcpmesh::daemon::{MeshState, build_services, spawn_accept_loop};
use mcpmesh::pairing::rendezvous::{SelfBinding, redeem_invite};
use mcpmesh::pairing::sas::short_auth_code;
use mcpmesh::pairing::{Invite, LiveInvites};
use mcpmesh::roster::gate::RosterGate;
use mcpmesh_net::framing::{FrameReader, Inbound, write_frame};
use mcpmesh_net::registry::ConnRegistry;
use mcpmesh_net::{ALPN_MCP, ALPN_PAIR, TrustGate, connect};
use serde_json::{Value, json};
use tokio::io::BufReader;
use tokio::time::timeout;
const MAX_PAIR_FRAME: usize = 64 * 1024;
const STUB: &str = env!("CARGO_BIN_EXE_echo_mcp_stub");
async fn inviter_endpoint() -> iroh::Endpoint {
iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![ALPN_MCP.to_vec(), ALPN_PAIR.to_vec()])
.bind()
.await
.expect("bind inviter endpoint")
}
async fn redeemer_endpoint() -> iroh::Endpoint {
iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![ALPN_MCP.to_vec()])
.bind()
.await
.expect("bind redeemer endpoint")
}
fn make_invite(
secret: [u8; 32],
inviter_id: [u8; 32],
services: &[&str],
expires_at_epoch: u64,
) -> Invite {
Invite {
secret,
inviter_id,
inviter_addr_json: "{}".into(), nickname: "alice".into(),
services: services.iter().map(|s| s.to_string()).collect(),
expires_at_epoch,
app_label: None,
}
}
fn hello_frame(secret: &[u8; 32], redeemer_id: &[u8; 32], nickname: &str) -> Value {
json!({
"secret": secret.to_vec(),
"redeemer_id": redeemer_id.to_vec(),
"redeemer_nickname": nickname,
})
}
async fn drive_redeemer(
redeemer: &iroh::Endpoint,
addr: iroh::EndpointAddr,
hello: Value,
) -> Value {
let conn = redeemer
.connect(addr, ALPN_PAIR)
.await
.expect("dial pair/1");
let (mut send, recv) = conn.open_bi().await.expect("open bi-stream");
write_frame(&mut send, &hello).await.expect("send hello");
let _ = send.finish();
let mut reader = FrameReader::new(BufReader::new(recv), MAX_PAIR_FRAME);
let reply = match reader.next().await.expect("read reply frame") {
Some(Inbound::Frame(v)) => v,
other => panic!("expected a reply frame, got {other:?}"),
};
drop(conn);
reply
}
async fn setup() -> (
iroh::Endpoint, // redeemer
iroh::EndpointAddr,
Arc<PeerStore>,
Arc<LiveInvites>,
[u8; 32], // inviter id
) {
let (redeemer, addr, store, invites, inviter_id, _cfg) = setup_full("").await;
(redeemer, addr, store, invites, inviter_id)
}
async fn setup_full(
config_toml: &str,
) -> (
iroh::Endpoint, // redeemer
iroh::EndpointAddr,
Arc<PeerStore>,
Arc<LiveInvites>,
[u8; 32], // inviter id
std::path::PathBuf,
) {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("state.redb");
let config_path = dir.path().join("config.toml");
if !config_toml.is_empty() {
std::fs::write(&config_path, config_toml).unwrap();
}
let store = Arc::new(PeerStore::open(&db_path).unwrap());
std::mem::forget(dir);
let gate: Arc<dyn TrustGate> = Arc::new(AllowlistGate::new(store.clone()));
let invites = Arc::new(LiveInvites::new());
let inviter = inviter_endpoint().await;
let inviter_id = *inviter.id().as_bytes();
let addr = inviter.addr();
let cfg = Config::load(&config_path).unwrap();
let mesh = MeshState::new(
inviter,
gate,
store.clone(),
invites.clone(),
"alice".into(),
config_path.clone(),
Arc::new(RosterGate::empty()),
Arc::new(ConnRegistry::new()),
None,
None,
None,
None,
);
let task = spawn_accept_loop(mesh.clone(), Arc::new(build_services(&cfg)));
mesh.set_accept_task(task).await;
let redeemer = redeemer_endpoint().await;
(redeemer, addr, store, invites, inviter_id, config_path)
}
const FUTURE: u64 = 4_000_000_000;
const PAST: u64 = 1_000;
#[tokio::test]
async fn happy_path_writes_the_trust_grant_and_returns_the_inviter_identity() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id) = setup().await;
let redeemer_id = *redeemer.id().as_bytes();
let secret = [7u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes", "kb"], FUTURE));
assert_eq!(invites.count(), 1);
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "bob")).await;
assert_eq!(reply["result"], "ok", "expected Ok reply, got {reply}");
assert_eq!(reply["inviter_nickname"], "alice");
let reply_inviter_id: Vec<u64> = reply["inviter_id"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_u64().unwrap())
.collect();
assert_eq!(
reply_inviter_id,
inviter_id.iter().map(|b| *b as u64).collect::<Vec<_>>()
);
let entry = store
.resolve(&redeemer_id)
.unwrap()
.expect("a PeerEntry must be written for the redeemer");
assert_eq!(entry.nickname, "bob");
assert!(
entry.services.is_empty(),
"the inviter's dial-back entry carries no service grants (§4.2): {:?}",
entry.services
);
assert!(entry.paired_at.is_some(), "pairing must stamp paired_at");
assert_eq!(invites.count(), 0, "a successful redeem burns the invite");
let inviter_sas = short_auth_code(&inviter_id, &redeemer_id, &secret);
let redeemer_sas = short_auth_code(&redeemer_id, &inviter_id, &secret);
assert_eq!(
inviter_sas, redeemer_sas,
"SAS must be endpoint-order-independent"
);
assert_eq!(inviter_sas.split('-').count(), 3, "SAS is three words");
})
.await
.expect("happy-path test timed out");
}
#[tokio::test]
async fn wrong_secret_is_refused_and_leaves_the_live_invite_untouched() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id) = setup().await;
let redeemer_id = *redeemer.id().as_bytes();
invites.mint(make_invite([1u8; 32], inviter_id, &["notes"], FUTURE));
assert_eq!(invites.count(), 1);
let reply = drive_redeemer(
&redeemer,
addr,
hello_frame(&[2u8; 32], &redeemer_id, "bob"),
)
.await;
assert_eq!(reply["result"], "refused", "wrong secret must be refused");
assert_eq!(reply["reason"], "pairing refused");
assert!(
reply.get("inviter_id").is_none(),
"a refusal leaks no inviter id"
);
assert!(store.resolve(&redeemer_id).unwrap().is_none());
assert_eq!(
invites.count(),
1,
"an unknown secret must not burn the live invite"
);
})
.await
.expect("wrong-secret test timed out");
}
#[tokio::test]
async fn expired_invite_is_refused_and_writes_no_entry() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id) = setup().await;
let redeemer_id = *redeemer.id().as_bytes();
let secret = [3u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], PAST));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "bob")).await;
assert_eq!(
reply["result"], "refused",
"an expired invite must be refused"
);
assert_eq!(reply["reason"], "pairing refused");
assert!(store.resolve(&redeemer_id).unwrap().is_none());
})
.await
.expect("expired test timed out");
}
#[tokio::test]
async fn id_mismatch_is_refused_writes_no_entry_and_logs_no_peer_id() {
timeout(Duration::from_secs(60), async {
let log_buf = global_log_buf();
let (redeemer, addr, store, invites, inviter_id) = setup().await;
let redeemer_id = *redeemer.id().as_bytes();
let secret = [5u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], FUTURE));
let fake_id = [9u8; 32];
assert_ne!(
fake_id, redeemer_id,
"the fake id must differ from the real TLS id"
);
let reply = drive_redeemer(&redeemer, addr, hello_frame(&secret, &fake_id, "bob")).await;
assert_eq!(reply["result"], "refused", "an id mismatch must be refused");
assert_eq!(reply["reason"], "id mismatch");
assert!(store.resolve(&redeemer_id).unwrap().is_none());
assert!(store.resolve(&fake_id).unwrap().is_none());
assert_eq!(
invites.count(),
1,
"an id mismatch must not touch the invite"
);
let logs = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap();
let ours: String = logs
.lines()
.filter(|l| l.contains("mcpmesh_node::"))
.collect::<Vec<_>>()
.join("\n");
assert!(
ours.contains("pair attempt refused: id mismatch"),
"our id-mismatch refusal must be logged: {logs:?}"
);
let real_b32 = iroh::EndpointId::from_bytes(&redeemer_id)
.unwrap()
.to_string();
assert!(
!ours.contains(&real_b32),
"the redeemer's EndpointId must never appear in our logs: {ours:?}"
);
})
.await
.expect("id-mismatch test timed out");
}
#[tokio::test]
async fn malformed_hello_is_refused_without_panicking_and_writes_no_entry() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id) = setup().await;
let redeemer_id = *redeemer.id().as_bytes();
invites.mint(make_invite([4u8; 32], inviter_id, &["notes"], FUTURE));
let conn = redeemer
.connect(addr, ALPN_PAIR)
.await
.expect("dial pair/1");
let (mut send, recv) = conn.open_bi().await.expect("open bi-stream");
send.write_all(b"this is not a valid hello at all\n")
.await
.expect("write garbage");
let _ = send.finish();
let mut reader = FrameReader::new(BufReader::new(recv), MAX_PAIR_FRAME);
let reply = match reader.next().await.expect("read reply frame") {
Some(Inbound::Frame(v)) => v,
other => panic!("expected a refusal frame, got {other:?}"),
};
drop(conn);
assert_eq!(reply["result"], "refused", "garbage must be refused");
assert_eq!(reply["reason"], "malformed request");
assert!(store.resolve(&redeemer_id).unwrap().is_none());
assert_eq!(
invites.count(),
1,
"a malformed hello must not touch the invite"
);
})
.await
.expect("malformed-hello test timed out");
}
fn seed_peer(store: &PeerStore, endpoint_id: [u8; 32], nickname: &str, services: &[&str]) {
store
.add(PeerEntry {
endpoint_id,
nickname: nickname.into(),
services: services.iter().map(|s| s.to_string()).collect(),
paired_at: None,
user_id: None,
last_addr: None,
})
.unwrap();
}
#[tokio::test]
async fn collision_guard_allows_a_fresh_unique_nickname() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, config_path) =
setup_full(&format!("[services.notes]\nrun = ['{STUB}']\nallow = []\n")).await;
let redeemer_id = *redeemer.id().as_bytes();
let secret = [21u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], FUTURE));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "bob")).await;
assert_eq!(
reply["result"], "ok",
"a fresh unique name must pair: {reply}"
);
let entry = store
.resolve(&redeemer_id)
.unwrap()
.expect("bob entry written");
assert_eq!(entry.nickname, "bob");
let cfg = Config::load(&config_path).unwrap();
assert_eq!(
cfg.services.get("notes").unwrap().allow,
vec!["bob".to_string()]
);
})
.await
.expect("fresh-unique test timed out");
}
#[tokio::test]
async fn collision_guard_allows_same_peer_re_pair() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, _cfg) = setup_full(&format!(
"[services.notes]\nrun = ['{STUB}']\nallow = [\"bob\"]\n"
))
.await;
let redeemer_id = *redeemer.id().as_bytes();
seed_peer(&store, redeemer_id, "bob", &[]);
let secret = [22u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], FUTURE));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "bob")).await;
assert_eq!(
reply["result"], "ok",
"the same peer re-pairing to the same service must be allowed: {reply}"
);
assert_eq!(invites.count(), 0, "the invite is consumed");
})
.await
.expect("same-peer re-pair test timed out");
}
#[tokio::test]
async fn collision_guard_allows_same_peer_additional_service() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, config_path) = setup_full(&format!(
"[services.notes]\nrun = ['{STUB}']\nallow = [\"bob\"]\n\
[services.kb]\nrun = ['{STUB}']\nallow = []\n"
))
.await;
let redeemer_id = *redeemer.id().as_bytes();
seed_peer(&store, redeemer_id, "bob", &["notes"]);
let secret = [23u8; 32];
invites.mint(make_invite(secret, inviter_id, &["kb"], FUTURE));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "bob")).await;
assert_eq!(
reply["result"], "ok",
"the same peer pairing to an additional service must be allowed: {reply}"
);
let cfg = Config::load(&config_path).unwrap();
assert_eq!(
cfg.services.get("kb").unwrap().allow,
vec!["bob".to_string()]
);
assert_eq!(
cfg.services.get("notes").unwrap().allow,
vec!["bob".to_string()]
);
})
.await
.expect("additional-service re-pair test timed out");
}
#[tokio::test]
async fn collision_guard_refuses_impersonating_an_existing_peer() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, config_path) =
setup_full(&format!("[services.notes]\nrun = ['{STUB}']\nallow = []\n")).await;
let redeemer_id = *redeemer.id().as_bytes();
let carol_id = [0xC0u8; 32];
assert_ne!(carol_id, redeemer_id);
seed_peer(&store, carol_id, "carol", &["notes"]);
let secret = [24u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], FUTURE));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "carol")).await;
assert_eq!(
reply["result"], "refused",
"impersonation must be refused: {reply}"
);
assert_eq!(reply["reason"], "pairing refused");
assert!(
store.resolve(&redeemer_id).unwrap().is_none(),
"no entry may be written under the impersonator's id"
);
let cfg = Config::load(&config_path).unwrap();
assert!(
cfg.services.get("notes").unwrap().allow.is_empty(),
"no grant may be applied for an impersonation attempt"
);
assert_eq!(invites.count(), 0, "the burned invite is not preserved");
})
.await
.expect("impersonation test timed out");
}
#[tokio::test]
async fn collision_guard_refuses_an_orphan_allow_name() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, _cfg) = setup_full(&format!(
"[services.notes]\nrun = ['{STUB}']\nallow = [\"carol\"]\n"
))
.await;
let redeemer_id = *redeemer.id().as_bytes();
let secret = [25u8; 32];
invites.mint(make_invite(secret, inviter_id, &["notes"], FUTURE));
let reply =
drive_redeemer(&redeemer, addr, hello_frame(&secret, &redeemer_id, "carol")).await;
assert_eq!(
reply["result"], "refused",
"an orphan-allow name must be refused: {reply}"
);
assert_eq!(reply["reason"], "pairing refused");
assert!(
store.resolve(&redeemer_id).unwrap().is_none(),
"no entry may be written for an orphan-allow claim"
);
assert_eq!(invites.count(), 0, "the burned invite is not preserved");
})
.await
.expect("orphan-allow test timed out");
}
#[tokio::test]
async fn reverse_pairing_preserves_the_inviters_dial_directory_and_nickname() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, store, invites, inviter_id, config_path) =
setup_full(&format!("[services.code]\nrun = ['{STUB}']\nallow = []\n")).await;
let alice_id = *redeemer.id().as_bytes();
store
.add(PeerEntry {
endpoint_id: alice_id,
nickname: "alice".into(),
services: vec!["notes".into()],
paired_at: Some("1000".into()),
user_id: Some("b64u:ALICE".into()),
last_addr: None,
})
.unwrap();
let secret = [31u8; 32];
invites.mint(make_invite(secret, inviter_id, &["code"], FUTURE));
let reply = drive_redeemer(
&redeemer,
addr,
hello_frame(&secret, &alice_id, "alice-laptop"),
)
.await;
assert_eq!(
reply["result"], "ok",
"the reverse pairing must succeed: {reply}"
);
let entry = store
.resolve(&alice_id)
.unwrap()
.expect("the alice entry survives the reverse pairing");
assert_eq!(
entry.nickname, "alice",
"the inviter's chosen nickname must not be renamed by the redeemer's self-suggestion"
);
assert_eq!(
entry.services,
vec!["notes".to_string()],
"the inviter's dial directory must be preserved, not wiped to []"
);
assert_eq!(
entry.user_id.as_deref(),
Some("b64u:ALICE"),
"a verified user_id is never downgraded to None by a binding-less re-pair"
);
assert_eq!(
entry.paired_at.as_deref(),
Some("1000"),
"the original pairing stamp is kept on the inviter side"
);
let cfg = Config::load(&config_path).unwrap();
assert_eq!(
cfg.services.get("code").unwrap().allow,
vec!["alice".to_string()],
"the grant must target the stored nickname, not the self-suggestion"
);
})
.await
.expect("reverse-pairing merge test timed out");
}
#[tokio::test]
async fn repeat_grant_unions_the_redeemers_dial_directory_and_applies_the_new_nickname() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, _store, invites, inviter_id, _cfg) = setup_full("").await;
let bob_dir = tempfile::tempdir().unwrap();
let bob_store = Arc::new(PeerStore::open(&bob_dir.path().join("state.redb")).unwrap());
bob_store
.add(PeerEntry {
endpoint_id: inviter_id,
nickname: "alice-old".into(),
services: vec!["notes".into()],
paired_at: Some("1000".into()),
user_id: Some("b64u:ALICE".into()),
last_addr: None,
})
.unwrap();
let secret = [32u8; 32];
let invite = Invite {
secret,
inviter_id,
inviter_addr_json: serde_json::to_string(&addr).unwrap(),
nickname: "alice".into(),
services: vec!["kb".into(), "notes".into()],
expires_at_epoch: FUTURE,
app_label: None,
};
invites.mint(invite.clone());
let result = redeem_invite(
redeemer,
"bob".into(),
invite.encode(),
bob_store.clone(),
None,
&bob_dir.path().join("config.toml"),
)
.await
.expect("the second redeem succeeds");
assert_eq!(result.peer_nickname, "alice");
let entry = bob_store
.resolve(&inviter_id)
.unwrap()
.expect("the alice entry survives the repeat grant");
assert_eq!(
entry.nickname, "alice",
"the NEW invite's suggested nickname is applied (rename-by-fresh-invite)"
);
assert_eq!(
entry.services,
vec!["notes".to_string(), "kb".to_string()],
"the dial directory UNIONs the repeat grant (dedup, existing order first)"
);
assert_eq!(
entry.user_id.as_deref(),
Some("b64u:ALICE"),
"a verified user_id is never clobbered to None by a binding-less re-pair"
);
assert_ne!(
entry.paired_at.as_deref(),
Some("1000"),
"the redeemer stamps each redeem as a fresh pairing event"
);
drop(bob_dir);
})
.await
.expect("repeat-grant union test timed out");
}
#[tokio::test]
async fn redeem_refuses_an_invite_squatting_an_existing_peers_nickname() {
timeout(Duration::from_secs(60), async {
let (redeemer, addr, _store, invites, mallory_id, _cfg) = setup_full("").await;
let bob_dir = tempfile::tempdir().unwrap();
let bob_store = Arc::new(PeerStore::open(&bob_dir.path().join("state.redb")).unwrap());
let real_alice_id = [0xA1u8; 32];
assert_ne!(real_alice_id, mallory_id);
bob_store
.add(PeerEntry {
endpoint_id: real_alice_id,
nickname: "alice".into(),
services: vec!["notes".into()],
paired_at: Some("1000".into()),
user_id: Some("b64u:ALICE".into()),
last_addr: None,
})
.unwrap();
let secret = [33u8; 32];
let invite = Invite {
secret,
inviter_id: mallory_id,
inviter_addr_json: serde_json::to_string(&addr).unwrap(),
nickname: "alice".into(),
services: vec!["kb".into()],
expires_at_epoch: FUTURE,
app_label: None,
};
invites.mint(invite.clone());
let err = redeem_invite(
redeemer,
"bob".into(),
invite.encode(),
bob_store.clone(),
None,
&bob_dir.path().join("config.toml"),
)
.await
.expect_err("redeeming a nickname-squatting invite must fail");
let msg = err.to_string();
assert!(
msg.contains("alice"),
"the error must name the colliding nickname so the user can rename: {msg}"
);
let alice = bob_store
.resolve(&real_alice_id)
.unwrap()
.expect("the real alice entry survives a squatting attempt");
assert_eq!(alice.nickname, "alice");
assert_eq!(alice.services, vec!["notes".to_string()]);
assert!(
bob_store.resolve(&mallory_id).unwrap().is_none(),
"no entry may be written for the squatter"
);
drop(bob_dir);
})
.await
.expect("redeemer-side squatting test timed out");
}
#[tokio::test]
async fn paired_and_granted_peer_is_admitted_to_the_service_end_to_end() {
timeout(Duration::from_secs(60), async {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("state.redb");
let config_path = dir.path().join("config.toml");
std::fs::write(
&config_path,
format!("[services.notes]\nrun = ['{STUB}']\nallow = []\n"),
)
.unwrap();
let store = Arc::new(PeerStore::open(&db_path).unwrap());
let gate: Arc<dyn TrustGate> = Arc::new(AllowlistGate::new(store.clone()));
let invites = Arc::new(LiveInvites::new());
let alice = inviter_endpoint().await;
let alice_id = *alice.id().as_bytes();
let alice_addr = alice.addr();
let cfg = Config::load(&config_path).unwrap();
let mesh = MeshState::new(
alice,
gate,
store.clone(),
invites.clone(),
"alice".into(),
config_path.clone(),
Arc::new(RosterGate::empty()),
Arc::new(ConnRegistry::new()),
None,
None,
None,
None,
);
let task = spawn_accept_loop(mesh.clone(), Arc::new(build_services(&cfg)));
mesh.set_accept_task(task).await;
let secret = [11u8; 32];
let invite = Invite {
secret,
inviter_id: alice_id,
inviter_addr_json: serde_json::to_string(&alice_addr).unwrap(),
nickname: "alice".into(),
services: vec!["notes".into()],
expires_at_epoch: FUTURE,
app_label: None,
};
invites.mint(invite.clone());
let bob = redeemer_endpoint().await;
let bob_id = *bob.id().as_bytes();
let bob_dir = tempfile::tempdir().unwrap();
let bob_store = Arc::new(PeerStore::open(&bob_dir.path().join("state.redb")).unwrap());
let result = redeem_invite(
bob.clone(),
"bob".into(),
invite.encode(),
bob_store.clone(),
None,
&bob_dir.path().join("config.toml"),
)
.await
.expect("redeem_invite dials, verifies the inviter id, sends the secret, succeeds");
assert_eq!(result.peer_nickname, "alice");
assert_eq!(result.services, vec!["notes".to_string()]);
let bob_side = bob_store
.resolve(&alice_id)
.unwrap()
.expect("bob's store has an alice entry");
assert_eq!(bob_side.nickname, "alice");
assert_eq!(bob_side.services, vec!["notes".to_string()]);
assert!(bob_side.paired_at.is_some());
let alice_side = store
.resolve(&bob_id)
.unwrap()
.expect("alice's store has a bob entry");
assert_eq!(alice_side.nickname, "bob");
assert!(
alice_side.services.is_empty(),
"alice's dial-back entry carries no service grants (§4.2): {:?}",
alice_side.services
);
assert!(alice_side.paired_at.is_some());
let after = Config::load(&config_path).unwrap();
assert_eq!(
after.services.get("notes").unwrap().allow,
vec!["bob".to_string()],
"the pairing grant must append bob to [services.notes].allow"
);
let expected_sas = short_auth_code(&alice_id, &bob_id, &secret);
assert_eq!(
result.sas_code, expected_sas,
"redeemer SAS must be correct"
);
assert_eq!(
short_auth_code(&bob_id, &alice_id, &secret),
expected_sas,
"SAS must be endpoint-order-independent (both sides read the same words)"
);
let mut transport = connect(&bob, alice_addr, "notes").await.unwrap();
transport
.send_value(json!({
"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {
"protocolVersion": "2025-06-18",
"_meta": {"mcpmesh/service": "notes"},
"capabilities": {}, "clientInfo": {"name": "bob", "version": "0"}
}
}))
.await
.unwrap();
let init = transport.recv_value().await.unwrap().unwrap();
assert_eq!(
init["result"]["serverInfo"]["name"], "echo-stub",
"the paired+granted peer must be ADMITTED to notes (not -32054'd): {init}"
);
transport
.send_value(json!({
"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": "echo", "arguments": {"text": "paired!"}}
}))
.await
.unwrap();
let call = transport.recv_value().await.unwrap().unwrap();
assert_eq!(call["result"]["content"][0]["text"], "paired!");
assert_eq!(
call["result"]["peer_name"], "bob",
"the gate-resolved identity 'bob' was injected into the served child"
);
drop(bob_dir);
std::mem::forget(dir);
})
.await
.expect("E2E admission test timed out");
}
#[tokio::test]
async fn pairing_exchanges_and_stores_each_sides_verified_user_id() {
timeout(Duration::from_secs(60), async {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("state.redb");
let config_path = dir.path().join("config.toml");
let store = Arc::new(PeerStore::open(&db_path).unwrap());
let gate: Arc<dyn TrustGate> = Arc::new(AllowlistGate::new(store.clone()));
let invites = Arc::new(LiveInvites::new());
let alice = inviter_endpoint().await;
let alice_id = *alice.id().as_bytes();
let alice_addr = alice.addr();
let (alice_uk, _) =
mcpmesh_trust::UserKey::load_or_generate(&dir.path().join("alice-user.key")).unwrap();
let alice_user_id = mcpmesh_trust::binding::user_id(&alice_uk);
let (a_pk, a_sig) = mcpmesh_trust::binding::present(&alice_uk, &alice_id);
let cfg = Config::load(&config_path).unwrap();
let mesh = MeshState::new(
alice,
gate,
store.clone(),
invites.clone(),
"alice".into(),
config_path.clone(),
Arc::new(RosterGate::empty()),
Arc::new(ConnRegistry::new()),
None,
None,
None,
None,
);
mesh.set_self_binding(Some(SelfBinding {
user_pk: a_pk,
sig: a_sig,
}));
let task = spawn_accept_loop(mesh.clone(), Arc::new(build_services(&cfg)));
mesh.set_accept_task(task).await;
let secret = [21u8; 32];
let invite = Invite {
secret,
inviter_id: alice_id,
inviter_addr_json: serde_json::to_string(&alice_addr).unwrap(),
nickname: "alice".into(),
services: vec![],
expires_at_epoch: FUTURE,
app_label: None,
};
invites.mint(invite.clone());
let bob = redeemer_endpoint().await;
let bob_id = *bob.id().as_bytes();
let bob_dir = tempfile::tempdir().unwrap();
let bob_store = Arc::new(PeerStore::open(&bob_dir.path().join("state.redb")).unwrap());
let (bob_uk, _) =
mcpmesh_trust::UserKey::load_or_generate(&bob_dir.path().join("bob-user.key")).unwrap();
let bob_user_id = mcpmesh_trust::binding::user_id(&bob_uk);
let (b_pk, b_sig) = mcpmesh_trust::binding::present(&bob_uk, &bob_id);
let pair_result = redeem_invite(
bob.clone(),
"bob".into(),
invite.encode(),
bob_store.clone(),
Some(SelfBinding {
user_pk: b_pk,
sig: b_sig,
}),
&bob_dir.path().join("config.toml"),
)
.await
.expect("redeem succeeds and exchanges self-sovereign bindings");
assert_eq!(
pair_result.peer_user_id.as_deref(),
Some(alice_user_id.as_str()),
"pair result must return alice's proven user_id to the redeemer"
);
let bob_side = bob_store
.resolve(&alice_id)
.unwrap()
.expect("bob has an alice entry");
assert_eq!(
bob_side.user_id.as_deref(),
Some(alice_user_id.as_str()),
"bob must store alice's proven self-sovereign user_id"
);
let alice_side = store
.resolve(&bob_id)
.unwrap()
.expect("alice has a bob entry");
assert_eq!(
alice_side.user_id.as_deref(),
Some(bob_user_id.as_str()),
"alice must store bob's proven self-sovereign user_id"
);
drop(bob_dir);
std::mem::forget(dir);
})
.await
.expect("user_id exchange test timed out");
}
#[tokio::test]
async fn redeem_refuses_an_address_swap_and_writes_no_entry_p3() {
timeout(Duration::from_secs(60), async {
let mallory = inviter_endpoint().await;
let mallory_id = *mallory.id().as_bytes();
let mallory_addr = mallory.addr();
tokio::spawn(async move {
while let Some(inc) = mallory.accept().await {
tokio::spawn(async move {
if let Ok(conn) = inc.await {
let _ = conn.accept_bi().await;
}
});
}
});
let named_inviter_id = [0xABu8; 32];
assert_ne!(
named_inviter_id, mallory_id,
"the named inviter id must differ from the dialed endpoint's real id"
);
let invite = Invite {
secret: [7u8; 32],
inviter_id: named_inviter_id,
inviter_addr_json: serde_json::to_string(&mallory_addr).unwrap(),
nickname: "alice".into(),
services: vec!["notes".into()],
expires_at_epoch: FUTURE,
app_label: None,
};
let bob = redeemer_endpoint().await;
let bob_dir = tempfile::tempdir().unwrap();
let bob_store = Arc::new(PeerStore::open(&bob_dir.path().join("state.redb")).unwrap());
let err = redeem_invite(
bob,
"bob".into(),
invite.encode(),
bob_store.clone(),
None,
&bob_dir.path().join("config.toml"),
)
.await
.expect_err("a P3 id mismatch must fail the redeem");
assert!(
err.to_string().contains("address-swap") || err.to_string().contains("id mismatch"),
"expected an address-swap / id-mismatch error, got: {err}"
);
assert!(bob_store.resolve(&named_inviter_id).unwrap().is_none());
assert!(bob_store.resolve(&mallory_id).unwrap().is_none());
drop(bob_dir);
})
.await
.expect("P3 negative test timed out");
}
fn global_log_buf() -> Arc<Mutex<Vec<u8>>> {
static LOG_BUF: OnceLock<Arc<Mutex<Vec<u8>>>> = OnceLock::new();
LOG_BUF
.get_or_init(|| {
let buf = Arc::new(Mutex::new(Vec::new()));
let subscriber = tracing_subscriber::fmt()
.with_writer(BufMakeWriter(buf.clone()))
.with_max_level(tracing::Level::INFO)
.with_ansi(false) .with_target(true) .without_time()
.finish();
let _ = tracing::subscriber::set_global_default(subscriber);
buf
})
.clone()
}
#[derive(Clone)]
struct BufMakeWriter(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for BufMakeWriter {
fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(data);
Ok(data.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for BufMakeWriter {
type Writer = BufMakeWriter;
fn make_writer(&'a self) -> Self::Writer {
self.clone()
}
}