use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
use mcpmesh_net::SessionTransport;
use mcpmesh_net::framing::{FrameReader, Inbound, write_frame};
use serde_json::Value;
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
use super::MeshState;
use crate::allowlist::PeerEntry;
async fn refuse_if_revoked(mesh: &Arc<MeshState>, id: [u8; 32], peer: &str) -> Result<()> {
let m = mesh.clone();
if crate::util::blocking("join dial revocation check", move || dial_refused(&m, &id)).await? {
return Err(revoked_refusal(peer));
}
Ok(())
}
fn revoked_refusal(peer: &str) -> anyhow::Error {
anyhow::anyhow!(
"{peer} is REVOKED on this node — dialling it would hand the request to a device you \
declared compromised. Use `mcpmesh revoke undo` if that was a mistake"
)
}
pub(crate) fn dial_refused(mesh: &MeshState, id: &[u8; 32]) -> bool {
refused_by(&mesh.store, mesh.roster.view().as_deref(), id)
}
pub(crate) fn refused_by(
store: &crate::allowlist::PeerStore,
view: Option<&mcpmesh_trust::roster::validate::RosterView>,
id: &[u8; 32],
) -> bool {
if store.is_refused(id) {
return true;
}
let Some(view) = view else {
return false;
};
view.is_revoked(id)
|| view
.resolve(id)
.is_some_and(|d| store.is_roster_user_revoked(&d.user_id))
}
pub async fn dial_service(
mesh: &Arc<MeshState>,
peer: &str,
service: &str,
) -> Result<SessionTransport> {
dial_service_with_idle_timeout(mesh, peer, service, None).await
}
fn warn_if_per_session_dropped(
per_conn: &Option<iroh::endpoint::QuicTransportConfig>,
peer: &str,
why: &str,
) {
if per_conn.is_some() {
tracing::warn!(
peer,
"idle_timeout_secs was IGNORED: {why} resolves to a racing dial, which opens several \
connections and keeps the winner. Name one device with eid: to apply it"
);
}
}
pub(crate) fn per_session_transport_config(
secs: Option<u64>,
keepalive_secs: u64,
) -> Result<Option<iroh::endpoint::QuicTransportConfig>> {
let idle = match secs {
None => return Ok(None),
Some(0) => None,
Some(s) => {
anyhow::ensure!(
s > keepalive_secs,
"idle_timeout_secs ({s}) must be greater than this node's keepalive interval \
({keepalive_secs}s): a keepalive arriving after the idle timer has fired would \
sever the session on a clock, even against a peer that is alive and answering"
);
Some(
iroh::endpoint::IdleTimeout::try_from(std::time::Duration::from_secs(s)).map_err(
|e| {
anyhow::anyhow!(
"idle_timeout_secs {s} is out of the range QUIC can encode: {e}"
)
},
)?,
)
}
};
let d = std::time::Duration::from_secs(keepalive_secs);
Ok(Some(
iroh::endpoint::QuicTransportConfig::builder()
.max_idle_timeout(idle)
.keep_alive_interval(d)
.default_path_keep_alive_interval(d)
.build(),
))
}
pub async fn dial_service_with_idle_timeout(
mesh: &Arc<MeshState>,
peer: &str,
service: &str,
idle_timeout_secs: Option<u64>,
) -> Result<SessionTransport> {
let keepalive = mesh.keep_alive_secs();
let per_conn = per_session_transport_config(idle_timeout_secs, keepalive)?;
if let Some(hex) = peer.strip_prefix("eid:") {
if let Ok(bytes) = data_encoding::HEXLOWER.decode(hex.as_bytes())
&& let Ok(id) = <[u8; 32]>::try_from(bytes.as_slice())
{
refuse_if_revoked(mesh, id, peer).await?;
}
return dial_by_eid(mesh, hex, service, per_conn).await;
}
if let Some(view) = mesh.roster.view() {
let devices = view.devices_for_user(peer);
if !devices.is_empty() {
let candidates = order_dial_candidates(&devices, &mesh.presence_table, peer);
warn_if_per_session_dropped(&per_conn, peer, "a roster person");
let candidates = hinted_addrs(mesh, candidates).await?;
let (transport, conn) = race_dial(&mesh.endpoint, candidates, service)
.await
.with_context(|| format!("dial {peer}/{service}"))?;
return Ok(watch_session(mesh, transport, conn));
}
}
let peer_owned = peer.to_string();
let store = mesh.store.clone();
let (single, multi): (Option<PeerEntry>, Vec<[u8; 32]>) =
tokio::task::spawn_blocking(move || -> Result<_> {
if let Some(e) = store.entry_for(&peer_owned)? {
return Ok((Some(e), Vec::new()));
}
let mut by_user = store.entries_for_user(&peer_owned)?;
match by_user.len() {
0 => Ok((None, Vec::new())),
1 => Ok((by_user.pop(), Vec::new())),
_ => Ok((None, by_user.iter().map(|e| e.endpoint_id).collect())),
}
})
.await
.context("join peer resolve")??;
if !multi.is_empty() {
warn_if_per_session_dropped(&per_conn, peer, "a user_id with several devices");
let multi = hinted_addrs(mesh, multi).await?;
let (transport, conn) = race_dial(&mesh.endpoint, multi, service)
.await
.with_context(|| format!("dial {peer}/{service}"))?;
return Ok(watch_session(mesh, transport, conn));
}
let entry = single.with_context(|| format!("peer '{peer}' is not in the allowlist"))?;
refuse_if_revoked(mesh, entry.endpoint_id, peer).await?;
let endpoint_id = iroh::EndpointId::from_bytes(&entry.endpoint_id)
.map_err(|e| anyhow::anyhow!("stored endpoint id for '{peer}' is invalid: {e}"))?;
let addr = stored_dial_addr(entry.last_addr.as_deref(), endpoint_id);
let (transport, conn) =
connect_with_timeout(&mesh.endpoint, addr, service, DIAL_TIMEOUT, per_conn)
.await
.with_context(|| format!("dial {peer}/{service}"))?;
Ok(watch_session(mesh, transport, conn))
}
fn watch_session(
mesh: &Arc<MeshState>,
transport: SessionTransport,
conn: iroh::endpoint::Connection,
) -> SessionTransport {
drop(super::path_watch::spawn(
mesh.clone(),
*conn.remote_id().as_bytes(),
&conn,
));
transport
}
async fn dial_by_eid(
mesh: &Arc<MeshState>,
hex: &str,
service: &str,
per_conn: Option<iroh::endpoint::QuicTransportConfig>,
) -> Result<SessionTransport> {
let bytes = data_encoding::HEXLOWER
.decode(hex.as_bytes())
.map_err(|_| anyhow::anyhow!("invalid eid principal: not lowercase hex"))?;
let endpoint_id_bytes: [u8; 32] = bytes
.as_slice()
.try_into()
.map_err(|_| anyhow::anyhow!("invalid eid principal: expected 32 bytes (64 hex chars)"))?;
let endpoint_id = iroh::EndpointId::from_bytes(&endpoint_id_bytes)
.map_err(|e| anyhow::anyhow!("invalid eid principal: {e}"))?;
let store = mesh.store.clone();
let last_addr = tokio::task::spawn_blocking(move || store.resolve(&endpoint_id_bytes))
.await
.context("join eid peer resolve")?
.ok()
.flatten()
.and_then(|e| e.last_addr);
let addr = stored_dial_addr(last_addr.as_deref(), endpoint_id);
let (transport, conn) =
connect_with_timeout(&mesh.endpoint, addr, service, DIAL_TIMEOUT, per_conn)
.await
.with_context(|| format!("dial eid:{hex}/{service}"))?;
Ok(watch_session(mesh, transport, conn))
}
pub(crate) async fn protocol_candidates(
mesh: &Arc<MeshState>,
peer: &str,
) -> anyhow::Result<Candidates> {
let resolved = resolve_candidates(mesh, peer).await?;
let mesh = mesh.clone();
crate::util::blocking("join dial revocation filter", move || {
let before = resolved.len();
let dialable: Vec<[u8; 32]> = resolved
.into_iter()
.filter(|id| !dial_refused(&mesh, id))
.collect();
Candidates {
refused: before - dialable.len(),
dialable,
}
})
.await
}
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Candidates {
pub(crate) dialable: Vec<[u8; 32]>,
pub(crate) refused: usize,
}
pub(crate) async fn connect_candidates(
mesh: &Arc<MeshState>,
peer: &str,
) -> anyhow::Result<Vec<[u8; 32]>> {
let c = protocol_candidates(mesh, peer).await?;
if c.dialable.is_empty() {
if c.refused > 0 {
return Err(revoked_refusal(peer));
}
anyhow::bail!("no peer '{peer}' — 'status' lists your peers and roster members");
}
Ok(c.dialable)
}
async fn resolve_candidates(mesh: &Arc<MeshState>, peer: &str) -> anyhow::Result<Vec<[u8; 32]>> {
if let Some(hex) = peer.strip_prefix("eid:") {
let bytes = data_encoding::HEXLOWER
.decode(hex.as_bytes())
.map_err(|_| anyhow::anyhow!("invalid eid principal: not lowercase hex"))?;
let id: [u8; 32] = bytes
.as_slice()
.try_into()
.map_err(|_| anyhow::anyhow!("invalid eid principal: expected 32 bytes"))?;
return Ok(vec![id]);
}
let mut out: Vec<[u8; 32]> = Vec::new();
if let Some(view) = mesh.roster.view() {
out.extend(view.devices_for_user(peer).into_iter().map(|(eid, _)| eid));
}
let store = mesh.store.clone();
let name = peer.to_string();
let from_store = crate::util::blocking("join connect_protocol resolve", move || {
let mut v: Vec<[u8; 32]> = Vec::new();
if let Some(e) = store.entry_for(&name)? {
v.push(e.endpoint_id);
}
v.extend(
store
.entries_for_user(&name)?
.into_iter()
.map(|e| e.endpoint_id),
);
Ok::<_, anyhow::Error>(v)
})
.await??;
out.extend(from_store);
let mut seen = std::collections::HashSet::new();
out.retain(|e| seen.insert(*e));
Ok(out)
}
pub(crate) async fn blob_source_addrs(
mesh: &Arc<MeshState>,
from: &[String],
) -> anyhow::Result<BlobSources> {
anyhow::ensure!(
from.len() <= mcpmesh_local_api::MAX_BLOB_SOURCES,
"too many blob sources: {} (the cap is {})",
from.len(),
mcpmesh_local_api::MAX_BLOB_SOURCES
);
let mut seen = std::collections::HashSet::new();
let mut out = Vec::new();
let mut skipped_revoked = 0usize;
for name in from {
let Candidates { dialable, refused } = protocol_candidates(mesh, name).await?;
anyhow::ensure!(
!dialable.is_empty() || refused > 0,
"no blob source '{name}' — it must be a paired peer, a roster member, or an \
`eid:`/`b64u:` principal"
);
if dialable.is_empty() {
skipped_revoked += 1;
}
for eid in dialable {
if !seen.insert(eid) {
continue;
}
let Ok(id) = iroh::EndpointId::from_bytes(&eid) else {
continue;
};
let store = mesh.store.clone();
let entry =
crate::util::blocking("join blob source resolve", move || store.resolve(&eid))
.await??;
out.push(stored_dial_addr(
entry.and_then(|e| e.last_addr).as_deref(),
id,
));
}
}
if skipped_revoked > 0 {
tracing::warn!(
skipped_revoked,
"blob sources skipped: REVOKED on this node, not dialled"
);
}
Ok(BlobSources {
addrs: out,
skipped_revoked,
})
}
#[derive(Debug)]
pub(crate) struct BlobSources {
pub(crate) addrs: Vec<iroh::EndpointAddr>,
pub(crate) skipped_revoked: usize,
}
pub(crate) fn stored_dial_addr(
last_addr: Option<&str>,
endpoint_id: iroh::EndpointId,
) -> iroh::EndpointAddr {
if let Some(json) = last_addr
&& let Ok(addr) = serde_json::from_str::<iroh::EndpointAddr>(json)
&& addr.id == endpoint_id
{
return dialable_only(addr);
}
iroh::EndpointAddr::from(endpoint_id)
}
pub(crate) fn dialable_only(addr: iroh::EndpointAddr) -> iroh::EndpointAddr {
let id = addr.id;
let kept: Vec<iroh::TransportAddr> = addr.addrs.into_iter().filter(is_dialable_addr).collect();
if kept.is_empty() {
iroh::EndpointAddr::from(id)
} else {
iroh::EndpointAddr::from_parts(id, kept)
}
}
pub(crate) fn is_dialable_addr(a: &iroh::TransportAddr) -> bool {
let iroh::TransportAddr::Ip(s) = a else {
return true;
};
if s.port() == 0 {
return false;
}
match s.ip().to_canonical() {
std::net::IpAddr::V4(v4) => {
!v4.is_unspecified()
&& !v4.is_multicast()
&& !v4.is_broadcast()
&& v4.octets()[0] < 240
}
std::net::IpAddr::V6(v6) => !v6.is_unspecified() && !v6.is_multicast(),
}
}
const DIAL_STAGGER: Duration = Duration::from_millis(500);
pub(crate) const DIAL_TIMEOUT: Duration = Duration::from_secs(20);
pub(crate) async fn connect_with_timeout(
endpoint: &iroh::Endpoint,
addr: iroh::EndpointAddr,
service: &str,
timeout: Duration,
transport: Option<iroh::endpoint::QuicTransportConfig>,
) -> Result<(SessionTransport, iroh::endpoint::Connection)> {
match tokio::time::timeout(
timeout,
mcpmesh_net::connect_with_transport_config(endpoint, addr, service, transport),
)
.await
{
Ok(r) => r.map_err(Into::into),
Err(_) => anyhow::bail!("dial timed out after {timeout:?}"),
}
}
fn order_dial_candidates(
devices: &[([u8; 32], String)],
presence: &crate::roster::presence::PresenceTable,
user_id: &str,
) -> Vec<[u8; 32]> {
let by_recency = presence.endpoints_for_user_by_recency(user_id);
let recency_rank = |eid: &[u8; 32]| -> usize {
by_recency
.iter()
.position(|e| e == eid)
.unwrap_or(usize::MAX)
};
let mut ordered: Vec<([u8; 32], String)> = devices.to_vec();
ordered.sort_by_key(|(eid, role)| (dial_role_rank(role), recency_rank(eid)));
ordered.into_iter().map(|(eid, _)| eid).collect()
}
pub(crate) fn dial_role_rank(role: &str) -> u8 {
match role {
"primary" => 0,
"mirror" => 1,
_ => 2,
}
}
pub async fn race_dial(
endpoint: &iroh::Endpoint,
candidates: Vec<iroh::EndpointAddr>,
service: &str,
) -> Result<(SessionTransport, iroh::endpoint::Connection)> {
anyhow::ensure!(!candidates.is_empty(), "no dial candidates to race");
let mut set: tokio::task::JoinSet<Result<(SessionTransport, iroh::endpoint::Connection)>> =
tokio::task::JoinSet::new();
let spawn_dial =
|set: &mut tokio::task::JoinSet<Result<(SessionTransport, iroh::endpoint::Connection)>>,
addr: iroh::EndpointAddr| {
let ep = endpoint.clone();
let svc = service.to_string();
set.spawn(async move { dial_one(&ep, addr, &svc).await });
};
let mut next = 0usize; spawn_dial(&mut set, candidates[next].clone()); next += 1;
let mut last_err: Option<anyhow::Error> = None;
loop {
if next < candidates.len() {
tokio::select! {
biased;
joined = set.join_next() => match joined {
Some(Ok(Ok(t))) => return Ok(t), Some(Ok(Err(e))) => last_err = Some(e), Some(Err(e)) => last_err = Some(anyhow::anyhow!("dial task join error: {e}")),
None => {
spawn_dial(&mut set, candidates[next].clone());
next += 1;
}
},
() = tokio::time::sleep(DIAL_STAGGER) => {
spawn_dial(&mut set, candidates[next].clone());
next += 1;
}
}
} else {
match set.join_next().await {
Some(Ok(Ok(t))) => return Ok(t),
Some(Ok(Err(e))) => last_err = Some(e),
Some(Err(e)) => last_err = Some(anyhow::anyhow!("dial task join error: {e}")),
None => {
return Err(
last_err.unwrap_or_else(|| anyhow::anyhow!("all dial candidates failed"))
);
}
}
}
}
}
async fn hinted_addrs(
mesh: &Arc<MeshState>,
candidates: Vec<[u8; 32]>,
) -> Result<Vec<iroh::EndpointAddr>> {
let mesh = mesh.clone();
crate::util::blocking("join dial-candidate hints", move || {
let candidates: Vec<[u8; 32]> = candidates
.into_iter()
.filter(|id| !dial_refused(&mesh, id))
.collect();
anyhow::ensure!(
!candidates.is_empty(),
"every device of that peer is REVOKED on this node"
);
let store = &mesh.store;
let mut out = Vec::with_capacity(candidates.len());
for eid in candidates {
let Ok(id) = iroh::EndpointId::from_bytes(&eid) else {
continue; };
let last = store.resolve(&eid).ok().flatten().and_then(|e| e.last_addr);
out.push(stored_dial_addr(last.as_deref(), id));
}
Ok(out)
})
.await?
}
async fn dial_one(
endpoint: &iroh::Endpoint,
addr: iroh::EndpointAddr,
service: &str,
) -> Result<(SessionTransport, iroh::endpoint::Connection)> {
connect_with_timeout(endpoint, addr, service, DIAL_TIMEOUT, None).await
}
pub async fn pipe_session<CR, CW>(
mut transport: SessionTransport,
service: &str,
mut control_reader: FrameReader<CR>,
mut control_writer: CW,
) -> Result<()>
where
CR: AsyncRead + Unpin + Send,
CW: AsyncWrite + Unpin + Send,
{
let init = match control_reader.next().await? {
Some(Inbound::Frame(v)) => inject_service(v, service),
Some(Inbound::Violation(_)) | None => return Ok(()),
};
transport
.send_value(init)
.await
.context("forward initialize to peer")?;
let transport_writer = transport.writer();
let to_mesh = async {
loop {
match control_reader.next().await {
Ok(Some(Inbound::Frame(frame))) => {
if transport_writer.send_value(frame).await.is_err() {
break; }
}
Ok(Some(Inbound::Violation(_))) => break,
Ok(None) | Err(_) => break, }
}
let _ = transport_writer.shutdown().await;
std::future::pending::<()>().await
};
let to_control = async {
while let Ok(Some(frame)) = transport.recv_value().await {
if write_frame(&mut control_writer, &frame).await.is_err() {
break; }
}
};
tokio::select! {
() = to_mesh => {}
() = to_control => {}
}
let _ = transport.shutdown().await;
let _ = control_writer.shutdown().await;
Ok(())
}
fn inject_service(mut frame: Value, service: &str) -> Value {
let Some(obj) = frame.as_object_mut() else {
return frame;
};
let params = obj
.entry("params")
.or_insert_with(|| Value::Object(Default::default()));
if !params.is_object() {
*params = Value::Object(Default::default());
}
let params = params.as_object_mut().expect("params set to object above");
let meta = params
.entry("_meta")
.or_insert_with(|| Value::Object(Default::default()));
if !meta.is_object() {
*meta = Value::Object(Default::default()); }
let meta = meta.as_object_mut().expect("meta set to object above");
for key in mcpmesh_net::service::SERVICE_KEYS {
meta.insert(key.into(), Value::String(service.to_string()));
}
frame
}
#[cfg(test)]
mod source_tests {
use super::blob_source_addrs;
use crate::daemon::testutil::hermetic_mesh;
async fn mesh_with_peers(
dir: &std::path::Path,
peers: &[(&str, [u8; 32], Option<&str>)],
) -> std::sync::Arc<crate::daemon::MeshState> {
let cfg = dir.join("config.toml");
std::fs::write(&cfg, "").unwrap();
let mesh = hermetic_mesh(cfg).await;
for (nick, eid, user) in peers {
mesh.store
.add(crate::allowlist::PeerEntry {
endpoint_id: *eid,
nickname: (*nick).into(),
services: vec![],
paired_at: None,
user_id: user.map(|u| u.to_string()),
last_addr: None,
})
.unwrap();
}
mesh
}
fn eid_of(seed: u8) -> [u8; 32] {
*iroh::SecretKey::from_bytes(&[seed; 32]).public().as_bytes()
}
#[tokio::test(flavor = "multi_thread")]
async fn blob_sources_resolve_dedupe_and_refuse_a_typo() {
let dir = tempfile::tempdir().unwrap();
let (a, b) = (eid_of(41), eid_of(42));
let mesh = mesh_with_peers(
dir.path(),
&[("alice", a, Some("b64u:alice-key")), ("bob", b, None)],
)
.await;
let out = blob_source_addrs(&mesh, &["alice".into()])
.await
.unwrap()
.addrs;
assert_eq!(out.len(), 1, "a paired nickname must resolve to its device");
assert_eq!(*out[0].id.as_bytes(), a);
let out = blob_source_addrs(&mesh, &["b64u:alice-key".into()])
.await
.unwrap()
.addrs;
assert_eq!(*out[0].id.as_bytes(), a, "a b64u: user_id must resolve");
let stranger = eid_of(43);
let out = blob_source_addrs(
&mesh,
&[format!("eid:{}", data_encoding::HEXLOWER.encode(&stranger))],
)
.await
.unwrap()
.addrs;
assert_eq!(*out[0].id.as_bytes(), stranger);
let out = blob_source_addrs(&mesh, &["bob".into(), "alice".into()])
.await
.unwrap()
.addrs;
assert_eq!(
out.iter().map(|x| *x.id.as_bytes()).collect::<Vec<_>>(),
vec![b, a],
"the caller's order must survive resolution"
);
let out = blob_source_addrs(&mesh, &["alice".into(), "b64u:alice-key".into()])
.await
.unwrap()
.addrs;
assert_eq!(out.len(), 1, "one device named twice must be dialled once");
let err = blob_source_addrs(&mesh, &["nobody".into()])
.await
.expect_err("an unresolvable source must be refused");
assert!(
format!("{err:#}").contains("no blob source 'nobody'"),
"{err:#}"
);
let many: Vec<String> = (0..mcpmesh_local_api::MAX_BLOB_SOURCES + 1)
.map(|_| "alice".to_string())
.collect();
let err = blob_source_addrs(&mesh, &many)
.await
.expect_err("over the cap must be refused");
assert!(
format!("{err:#}").contains("too many blob sources"),
"{err:#}"
);
}
fn revoke(mesh: &crate::daemon::MeshState, eid: [u8; 32]) {
mesh.store
.revoke(crate::allowlist::RevokedEntry {
endpoint_id: eid,
revoked_at: 1,
reason: None,
source: "local".into(),
signer_user_id: None,
issued_at: None,
})
.unwrap();
}
fn revoke_identity(mesh: &crate::daemon::MeshState, user_id: &str) {
mesh.store
.revoke_user(
user_id,
&crate::allowlist::RevokedEntry {
endpoint_id: [0u8; 32],
revoked_at: 1,
reason: None,
source: "local".into(),
signer_user_id: None,
issued_at: None,
},
)
.unwrap();
}
fn eid_principal(eid: [u8; 32]) -> String {
format!("eid:{}", data_encoding::HEXLOWER.encode(&eid))
}
#[tokio::test(flavor = "multi_thread")]
async fn connect_protocol_candidates_never_include_a_revoked_device() {
let dir = tempfile::tempdir().unwrap();
let (laptop, phone, carol) = (eid_of(51), eid_of(52), eid_of(53));
let mesh = mesh_with_peers(
dir.path(),
&[
("bob-laptop", laptop, Some("b64u:bob")),
("bob-phone", phone, Some("b64u:bob")),
("carol", carol, None),
],
)
.await;
assert_eq!(
super::connect_candidates(&mesh, "bob-laptop")
.await
.unwrap(),
vec![laptop]
);
revoke(&mesh, laptop);
for sel in ["bob-laptop".to_string(), eid_principal(laptop)] {
let e = super::connect_candidates(&mesh, &sel)
.await
.expect_err("a revoked device must never be a connect_protocol candidate");
assert!(
format!("{e:#}").contains(&format!("{sel} is REVOKED on this node")),
"{sel}: {e:#}"
);
}
assert_eq!(
super::connect_candidates(&mesh, "b64u:bob").await.unwrap(),
vec![phone],
"the revoked laptop is dropped, the phone stays"
);
assert_eq!(
super::connect_candidates(&mesh, "carol").await.unwrap(),
vec![carol]
);
let e = super::connect_candidates(&mesh, "nobody")
.await
.expect_err("nobody resolves");
let msg = format!("{e:#}");
assert!(
msg.contains("no peer 'nobody'") && !msg.contains("REVOKED"),
"{msg}"
);
revoke(&mesh, phone);
let e = super::connect_candidates(&mesh, "b64u:bob")
.await
.expect_err("every device revoked");
assert!(
format!("{e:#}").contains("b64u:bob is REVOKED on this node"),
"{e:#}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn blob_sources_drop_a_revoked_device() {
let dir = tempfile::tempdir().unwrap();
let (stolen, carol) = (eid_of(54), eid_of(55));
let mesh = mesh_with_peers(
dir.path(),
&[("stolen", stolen, None), ("carol", carol, None)],
)
.await;
revoke(&mesh, stolen);
let out = blob_source_addrs(&mesh, &["stolen".into(), "carol".into()])
.await
.unwrap()
.addrs;
assert_eq!(
out.iter().map(|a| *a.id.as_bytes()).collect::<Vec<_>>(),
vec![carol],
"a revoked source must never reach the fetch's dial set"
);
let out = blob_source_addrs(&mesh, &[eid_principal(stolen)])
.await
.expect("a revoked-only source is dropped, not an error")
.addrs;
assert!(out.is_empty(), "{out:?}");
}
#[tokio::test(flavor = "multi_thread")]
async fn peer_services_refuses_revoked_roster_devices_while_read_only_verbs_work() {
let dir = tempfile::tempdir().unwrap();
let (mallory, alice, gone) = (eid_of(71), eid_of(72), eid_of(73));
let mesh = mesh_with_peers(dir.path(), &[("gone", gone, None)]).await;
let state = crate::control::DaemonState::with_mesh("test", mesh.clone());
mesh.roster.install(roster_view(
&[(mallory, "b64u:mallory"), (alice, "alice")],
&[gone],
));
revoke_identity(&mesh, "b64u:mallory");
assert!(
!mesh.store.is_refused(&gone),
"fixture: the store alone admits `gone`"
);
for sel in [
"gone".to_string(),
eid_principal(gone),
eid_principal(mallory),
] {
let e = crate::daemon::handlers::peer_services(&state, sel.clone())
.await
.expect_err("peer_services dials, so it must refuse a revoked roster device");
assert!(
format!("{e:#}").contains("is REVOKED on this node"),
"{sel}: {e:#}"
);
}
let e = crate::daemon::handlers::peer_services(&state, eid_principal(alice))
.await
.expect_err("alice is not running");
assert!(!format!("{e:#}").contains("REVOKED"), "{e:#}");
crate::daemon::handlers::peer_diagnostics(&state, "gone")
.await
.expect("peer_diagnostics dials nothing, so a roster revocation must not block it");
crate::daemon::handlers::peer_hint_clear(&state, "gone")
.await
.expect("clearing a stored hint dials nothing, so it must not be blocked either");
}
async fn counting_ping_peer() -> (
iroh::Endpoint,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
tokio::task::JoinHandle<()>,
) {
let ep = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![mcpmesh_net::ALPN_PING.to_vec()])
.bind()
.await
.unwrap();
let n = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let (e, c) = (ep.clone(), n.clone());
let task = tokio::spawn(async move {
while let Some(incoming) = e.accept().await {
if incoming.await.is_ok() {
c.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
}
});
(ep, n, task)
}
#[tokio::test(flavor = "multi_thread")]
async fn the_reachability_probe_never_dials_a_revoked_peer() {
tokio::time::timeout(std::time::Duration::from_secs(90), async {
let dir = tempfile::tempdir().unwrap();
let mesh = mesh_with_peers(dir.path(), &[]).await;
let (stolen_ep, stolen, _t1) = counting_ping_peer().await;
let (gone_ep, gone, _t2) = counting_ping_peer().await;
let (mal_ep, mal, _t3) = counting_ping_peer().await;
let (live_ep, live, _t4) = counting_ping_peer().await;
for (ep, nick) in [
(&stolen_ep, "stolen"),
(&gone_ep, "gone"),
(&mal_ep, "mal"),
(&live_ep, "live"),
] {
mesh.store
.add(crate::allowlist::PeerEntry {
endpoint_id: *ep.id().as_bytes(),
nickname: nick.into(),
services: vec![],
paired_at: None,
user_id: None,
last_addr: Some(serde_json::to_string(&ep.addr()).unwrap()),
})
.unwrap();
}
let id = |ep: &iroh::Endpoint| *ep.id().as_bytes();
revoke(&mesh, id(&stolen_ep));
mesh.roster.install(roster_view(
&[(id(&mal_ep), "b64u:mallory"), (id(&live_ep), "alice")],
&[id(&gone_ep)],
));
revoke_identity(&mesh, "b64u:mallory");
let refused = [
("stolen", id(&stolen_ep), &stolen),
("gone", id(&gone_ep), &gone),
("mal", id(&mal_ep), &mal),
];
for eid in refused.iter().map(|r| r.1).chain([id(&live_ep)]) {
mesh.reachability.lock().unwrap().insert(
eid,
crate::daemon::reach::ReachEntry {
reachable: true,
rtt_ms: Some(1),
probed_at: 1,
meta: String::new(),
services: vec![],
seq: 0,
observed: 0,
path: mcpmesh_local_api::PeerPath::Unknown,
},
);
}
let mut frames = mesh.reach_bcast.subscribe();
crate::daemon::reach::probe_peer(&mesh, id(&live_ep)).await;
for (_, eid, _) in &refused {
crate::daemon::reach::probe_peer(&mesh, *eid).await;
}
let _ = crate::daemon::reach::reachability_of(&mesh);
assert!(
live.load(std::sync::atomic::Ordering::SeqCst) > 0,
"control: a probe of an unrevoked peer must land"
);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
for (name, eid, count) in &refused {
assert_eq!(
count.load(std::sync::atomic::Ordering::SeqCst),
0,
"a revoked peer must never be probed ({name})"
);
let cache = mesh.reachability.lock().unwrap();
let row = cache.get(eid).expect("the old row is still there");
assert!(
row.reachable && row.probed_at == 1,
"a probe that never dialled must not freshen or flip the cached row ({name})"
);
}
let refused_principals: Vec<String> = refused
.iter()
.map(|r| mcpmesh_net::EndpointId::from_bytes(r.1).principal())
.collect();
let mut live_frames = 0;
while let Ok(f) = frames.try_recv() {
assert!(
!refused_principals
.iter()
.any(|p| f.peer.principal.as_deref() == Some(p.as_str())),
"no Probe frame may be sent for a probe that never went out: {:?}",
f.peer.principal
);
live_frames += 1;
}
assert!(
live_frames > 0,
"control: the live peer's real probe flips its row, so a frame IS sent for it"
);
let state = crate::control::DaemonState::with_mesh("test", mesh.clone());
let status = crate::control::status_result(&state).unwrap();
let source_of = |eid: [u8; 32]| {
let p = mcpmesh_net::EndpointId::from_bytes(eid).principal();
status
.revoked
.iter()
.find(|r| r.principal == p)
.map(|r| r.source.clone())
};
assert_eq!(source_of(id(&stolen_ep)).as_deref(), Some("local"));
assert_eq!(source_of(id(&gone_ep)).as_deref(), Some("roster"));
assert_eq!(source_of(id(&mal_ep)).as_deref(), Some("roster_identity"));
assert_eq!(
source_of(id(&live_ep)),
None,
"an unrevoked peer is not listed"
);
})
.await
.expect("probe test timed out");
}
#[tokio::test(flavor = "multi_thread")]
async fn the_gossip_bootstrap_excludes_refused_devices() {
let dir = tempfile::tempdir().unwrap();
let mesh = mesh_with_peers(dir.path(), &[]).await;
let (alice, mallory, bob, me) = (eid_of(81), eid_of(82), eid_of(83), eid_of(84));
let view = roster_view(
&[
(alice, "alice"),
(mallory, "b64u:mallory"),
(bob, "bob"),
(me, "me"),
],
&[],
);
let ids = |b: crate::daemon::boot::bootstrap::GossipBootstrap| {
let mut v: Vec<[u8; 32]> = b.ids().iter().map(|e| *e.as_bytes()).collect();
v.sort();
v
};
let mut everyone = vec![alice, mallory, bob];
everyone.sort();
assert_eq!(
ids(crate::daemon::boot::bootstrap::gossip_bootstrap(
&mesh.store,
Some(&view),
&me
)),
everyone,
"fixture: before any revocation every other device bootstraps"
);
revoke_identity(&mesh, "b64u:mallory");
revoke(&mesh, bob);
assert_eq!(
ids(crate::daemon::boot::bootstrap::gossip_bootstrap(
&mesh.store,
Some(&view),
&me
)),
vec![alice],
"gossip must not bootstrap from a device this node refuses to dial"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn the_roster_transport_plan_bootstraps_without_refused_devices() {
let dir = tempfile::tempdir().unwrap();
let mesh = mesh_with_peers(dir.path(), &[]).await;
let (alice, mallory, bob, me) = (eid_of(91), eid_of(92), eid_of(93), eid_of(94));
mesh.roster.install(roster_view(
&[
(alice, "alice"),
(mallory, "b64u:mallory"),
(bob, "bob"),
(me, "me"),
],
&[],
));
revoke_identity(&mesh, "b64u:mallory");
revoke(&mesh, bob);
let cfg = crate::config::Config::default();
let our_id = iroh::EndpointId::from_bytes(&me).unwrap();
assert!(
crate::daemon::boot::plan_roster_transport(
&mesh.roster,
&mesh.store,
&cfg,
false,
&our_id
)
.await
.is_none(),
"a pure-pairing daemon plans no transport"
);
let plan = crate::daemon::boot::plan_roster_transport(
&mesh.roster,
&mesh.store,
&cfg,
true,
&our_id,
)
.await
.expect("roster mode with an installed roster plans a transport");
assert_eq!(plan.org_id, "acme");
let ids: Vec<[u8; 32]> = plan.bootstrap.ids().iter().map(|e| *e.as_bytes()).collect();
assert_eq!(
ids,
vec![alice],
"the daemon must subscribe gossip with the filtered bootstrap set"
);
}
async fn mesh_with_roster_blobs(
dir: &std::path::Path,
) -> std::sync::Arc<crate::daemon::MeshState> {
use std::sync::Arc;
let store = Arc::new(crate::allowlist::PeerStore::open(&dir.join("state.redb")).unwrap());
let pairs = Arc::new(crate::allowlist::AllowlistGate::new(store.clone()));
let roster = Arc::new(crate::roster::gate::RosterGate::empty());
let gate: Arc<dyn mcpmesh_net::TrustGate> = Arc::new(
crate::roster::gate::ComposedGate::new(roster.clone(), pairs),
);
let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.bind()
.await
.unwrap();
let blobs = crate::roster::transport::RosterBlobs::new(&endpoint);
crate::daemon::MeshState::new(
endpoint,
gate,
store,
Arc::new(crate::pairing::LiveInvites::new()),
"test".into(),
dir.join("config.toml"),
roster,
Arc::new(mcpmesh_net::registry::ConnRegistry::new()),
None,
Some(blobs),
None,
None,
)
}
async fn counting_roster_provider() -> (
iroh::Endpoint,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
tokio::task::JoinHandle<()>,
) {
let ep = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![crate::roster::transport::BLOB_ALPN.to_vec()])
.bind()
.await
.unwrap();
let n = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let (e, c) = (ep.clone(), n.clone());
let task = tokio::spawn(async move {
while let Some(incoming) = e.accept().await {
if let Ok(conn) = incoming.await {
c.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
conn.close(0u32.into(), b"test");
}
}
});
(ep, n, task)
}
#[tokio::test(flavor = "multi_thread")]
async fn a_roster_announce_from_a_revoked_provider_is_not_fetched() {
use std::sync::atomic::Ordering;
tokio::time::timeout(std::time::Duration::from_secs(90), async {
let dir = tempfile::tempdir().unwrap();
let mesh = mesh_with_roster_blobs(dir.path()).await;
let book = std::sync::Arc::new(crate::roster::transport::RosterAddrBook::register(
&mesh.endpoint,
64,
));
assert!(mesh.roster_addr_book.set(book.clone()).is_ok());
let (stolen_ep, stolen, _t1) = counting_roster_provider().await;
let (gone_ep, gone, _t2) = counting_roster_provider().await;
let (mal_ep, mal, _t3) = counting_roster_provider().await;
let (live_ep, live, _t4) = counting_roster_provider().await;
let id = |ep: &iroh::Endpoint| *ep.id().as_bytes();
revoke(&mesh, id(&stolen_ep));
mesh.roster.install(roster_view(
&[(id(&mal_ep), "b64u:mallory")],
&[id(&gone_ep)],
));
revoke_identity(&mesh, "b64u:mallory");
let announce = |ep: &iroh::Endpoint| crate::roster::transport::RosterAnnounce {
serial: 99,
roster_hash: "blake3:00".into(),
blob_ticket: iroh_blobs::ticket::BlobTicket::new(
ep.addr(),
iroh_blobs::Hash::new(b"roster"),
iroh_blobs::BlobFormat::Raw,
)
.to_string(),
};
for ep in [&stolen_ep, &gone_ep, &mal_ep] {
crate::roster::distribute::on_announce(&mesh, announce(ep))
.await
.expect("a refused provider is a fail-safe no-op, not an error");
}
let (m, a) = (mesh.clone(), announce(&live_ep));
let control = tokio::spawn(async move {
let _ = crate::roster::distribute::on_announce(&m, a).await;
});
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
while live.load(Ordering::SeqCst) == 0 {
assert!(
tokio::time::Instant::now() < deadline,
"control: an unrevoked provider must be dialled"
);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
control.abort();
assert!(
book.recorded_for(&id(&live_ep)).is_some(),
"control: an unrevoked provider's address IS noted"
);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
for (name, ep, count) in [
("stolen", &stolen_ep, &stolen),
("gone", &gone_ep, &gone),
("mal", &mal_ep, &mal),
] {
assert_eq!(
count.load(Ordering::SeqCst),
0,
"a revoked roster-blob provider must never be contacted ({name})"
);
assert!(
book.recorded_for(&id(ep)).is_none(),
"a refused provider's address must not even be noted ({name})"
);
}
})
.await
.expect("announce test timed out");
}
fn roster_view(
active: &[([u8; 32], &str)],
revoked: &[[u8; 32]],
) -> mcpmesh_trust::roster::validate::RosterView {
use mcpmesh_trust::roster::{Roster, RosterDevice, RosterUser, encode_b64u};
let root = ed25519_dalek::SigningKey::from_bytes(&[9u8; 32]);
let users = active
.iter()
.map(|(eid, uid)| RosterUser {
user_id: (*uid).into(),
display_name: (*uid).into(),
user_pk: encode_b64u(&[1u8; 32]),
groups: vec!["team".into()],
devices: vec![RosterDevice {
endpoint_id: encode_b64u(eid),
label: "d".into(),
role: "primary".into(),
}],
})
.collect();
let r = mcpmesh_trust::roster::sign::mint_signed(
&root,
Roster {
format: "mcpmesh-roster/1".into(),
org_id: "acme".into(),
serial: 3,
issued_at: "2000-01-01T00:00:00Z".into(),
expires_at: "2999-01-01T00:00:00Z".into(),
groups: vec!["team".into()],
users,
revoked_endpoints: revoked.iter().map(|e| encode_b64u(e)).collect(),
successor_root_pk: None,
successor_sig: None,
sig: String::new(),
},
);
mcpmesh_trust::roster::validate::load_installed(&r, &root.verifying_key()).unwrap()
}
#[tokio::test(flavor = "multi_thread")]
async fn a_roster_device_under_a_revoked_b64u_identity_is_refused_outbound() {
let dir = tempfile::tempdir().unwrap();
let mesh = mesh_with_peers(dir.path(), &[]).await;
let (mallory, alice, gone) = (eid_of(61), eid_of(62), eid_of(63));
mesh.roster.install(roster_view(
&[(mallory, "b64u:mallory"), (alice, "alice")],
&[gone],
));
revoke_identity(&mesh, "b64u:mallory");
revoke_identity(&mesh, "alice");
assert!(mesh.store.resolve(&mallory).unwrap().is_none());
assert!(
!mesh.store.is_refused(&mallory),
"fixture: the store alone admits no refusal"
);
assert!(
mesh.gate.resolve(&mallory.into()).is_none(),
"fixture: rule 2 refuses inbound"
);
assert!(
mesh.gate.resolve(&alice.into()).is_some(),
"fixture: alice is admitted inbound"
);
const BOUND: std::time::Duration = std::time::Duration::from_secs(30);
let e =
match tokio::time::timeout(BOUND, super::dial_service(&mesh, "b64u:mallory", "notes"))
.await
.expect("the refusal is immediate, not a dial timeout")
{
Ok(_) => panic!("the race must refuse a device of a revoked roster identity"),
Err(e) => e,
};
assert!(
format!("{e:#}").contains("every device of that peer is REVOKED"),
"{e:#}"
);
for eid in [mallory, gone] {
let sel = eid_principal(eid);
let e = match tokio::time::timeout(BOUND, super::dial_service(&mesh, &sel, "notes"))
.await
.expect("immediate")
{
Ok(_) => panic!("an eid: dial must refuse {sel}"),
Err(e) => e,
};
assert!(
format!("{e:#}").contains(&format!("{sel} is REVOKED on this node")),
"{e:#}"
);
let e = super::connect_candidates(&mesh, &sel)
.await
.expect_err("connect_protocol must refuse it too");
assert!(
format!("{e:#}").contains("is REVOKED on this node"),
"{e:#}"
);
}
let e = super::connect_candidates(&mesh, "b64u:mallory")
.await
.expect_err("connect_protocol by roster person");
assert!(
format!("{e:#}").contains("is REVOKED on this node"),
"{e:#}"
);
assert_eq!(
super::connect_candidates(&mesh, "alice").await.unwrap(),
vec![alice]
);
for sel in ["alice".to_string(), eid_principal(alice)] {
if let Err(e) = tokio::time::timeout(BOUND, super::dial_service(&mesh, &sel, "notes"))
.await
.expect("a hermetic id-only dial fails fast")
{
assert!(
!format!("{e:#}").contains("REVOKED"),
"a live roster device must be dialled, not refused ({sel}): {e:#}"
);
}
}
assert!(mesh.store.unrevoke_user("b64u:mallory").unwrap());
assert_eq!(
super::connect_candidates(&mesh, "b64u:mallory")
.await
.unwrap(),
vec![mallory]
);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_v4_mapped_address_survives_the_hint_round_trip_as_v6() {
let id = iroh::SecretKey::from_bytes(&[8u8; 32]).public();
let mapped: std::net::SocketAddr = "[::ffff:224.0.0.1]:1900".parse().unwrap();
assert!(mapped.is_ipv6(), "precondition: Rust parses this as v6");
let json = serde_json::to_string(&iroh::EndpointAddr::from_parts(
id,
[iroh::TransportAddr::Ip(mapped)],
))
.unwrap();
let back: iroh::EndpointAddr = serde_json::from_str(&json).unwrap();
let iroh::TransportAddr::Ip(got) = back.addrs.iter().next().expect("one address") else {
panic!("expected an Ip transport address");
};
assert!(
got.is_ipv6(),
"the mapped form must survive serde as v6, or the bypass cases below test nothing: \
{got:?}"
);
assert!(
got.ip().to_canonical().is_ipv4(),
"and canonicalize to the v4 address iroh will actually send to: {got:?}"
);
}
#[test]
fn a_stored_hint_cannot_aim_a_dial_at_a_non_peer() {
let id = iroh::SecretKey::from_bytes(&[7u8; 32]).public();
let hint = |addrs: Vec<&str>| {
serde_json::to_string(&iroh::EndpointAddr::from_parts(
id,
addrs
.into_iter()
.map(|a| iroh::TransportAddr::Ip(a.parse().unwrap()))
.collect::<Vec<_>>(),
))
.unwrap()
};
for bad in [
"0.0.0.0:4433", "[::]:4433",
"224.0.0.1:1900", "239.255.255.250:1900",
"255.255.255.255:80", "[ff02::1]:4433", "[::ffff:0.0.0.0]:4433",
"[::ffff:224.0.0.1]:1900",
"[::ffff:239.255.255.250]:1900",
"[::ffff:255.255.255.255]:80",
"224.0.0.0:4433", "239.255.255.255:4433", "240.0.0.0:4433", "240.1.2.3:4433",
"1.2.3.4:0", "[::ffff:1.2.3.4]:0",
] {
let got = super::stored_dial_addr(Some(&hint(vec![bad])), id);
assert!(
got.addrs.is_empty(),
"`{bad}` can never be a QUIC peer and must not become a dial target: {got:?}"
);
assert_eq!(got.id, id, "and the dial still names the peer: {got:?}");
}
for good in [
"127.0.0.1:4433", "[::1]:4433",
"192.168.1.5:4433",
"[fe80::1]:4433", "169.254.3.4:4433", "1.2.3.4:4433",
"16.0.0.1:4433", "223.255.255.254:4433", "[::ffff:192.168.1.5]:4433", ] {
let got = super::stored_dial_addr(Some(&hint(vec![good])), id);
assert_eq!(
got.addrs.len(),
1,
"`{good}` is a legitimate peer address and must survive: {got:?}"
);
}
let mixed = hint(vec!["0.0.0.0:1", "192.168.1.5:4433", "224.0.0.1:2"]);
let got = super::stored_dial_addr(Some(&mixed), id);
assert_eq!(
got.addrs.len(),
1,
"only the dialable one survives: {got:?}"
);
let all_bad = hint(vec!["0.0.0.0:1", "224.0.0.1:2"]);
let got = super::stored_dial_addr(Some(&all_bad), id);
assert!(got.addrs.is_empty());
assert_eq!(got.id, id);
}
#[test]
fn inject_service_sets_both_meta_spellings_across_shapes() {
use serde_json::json;
let both = |f: &serde_json::Value, want: &str| {
for key in mcpmesh_net::service::SERVICE_KEYS {
assert_eq!(
f["params"]["_meta"][key], want,
"`{key}` must name the service: {f}"
);
}
};
let f = inject_service(json!({"method": "initialize"}), "kb");
both(&f, "kb");
assert_eq!(f["method"], "initialize");
let f = inject_service(json!({"params": {"x": 1}}), "loc");
assert_eq!(f["params"]["x"], 1);
both(&f, "loc");
both(&inject_service(json!({"params": 7}), "kb"), "kb");
both(
&inject_service(json!({"params": {"_meta": "nope"}}), "kb"),
"kb",
);
assert_eq!(inject_service(json!("scalar"), "kb"), json!("scalar"));
let f = inject_service(json!({"method": "initialize"}), "kb");
assert_eq!(
f["params"]["_meta"][mcpmesh_net::service::SERVICE_KEYS[0]],
f["params"]["_meta"][mcpmesh_net::service::SERVICE_KEYS[1]],
"the spellings must never diverge: {f}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn pipe_session_delivers_the_echo_after_control_eof() {
use mcpmesh_net::framing::{FrameReader, Inbound, write_frame};
use serde_json::json;
use tokio::io::duplex;
tokio::time::timeout(std::time::Duration::from_secs(20), async {
let server_ep = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![mcpmesh_net::ALPN_MCP.to_vec()])
.bind()
.await
.unwrap();
let server_addr = server_ep.addr();
let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
let peer = tokio::spawn(async move {
let incoming = server_ep.accept().await.expect("one inbound connection");
let conn = incoming.await.expect("handshake");
let (send, recv) = conn.accept_bi().await.expect("session bi-stream");
let mut t = mcpmesh_net::SessionTransport::new(recv, send, 1024 * 1024);
let mut seen = Vec::new();
while let Ok(Some(f)) = t.recv_value().await {
seen.push(f);
}
for f in &seen {
t.send_value(f.clone()).await.unwrap();
}
t.shutdown().await.unwrap(); let _ = done_rx.await; seen
});
let client_ep = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![mcpmesh_net::ALPN_MCP.to_vec()])
.bind()
.await
.unwrap();
let transport = mcpmesh_net::connect(&client_ep, server_addr, "echo")
.await
.unwrap()
.0;
let (mut ctl_in_w, ctl_in_r) = duplex(64 * 1024);
let (ctl_out_w, ctl_out_test_r) = duplex(64 * 1024);
let init = json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}});
write_frame(&mut ctl_in_w, &init).await.unwrap();
drop(ctl_in_w);
let session = tokio::spawn(pipe_session(
transport,
"echo",
FrameReader::new(ctl_in_r, 1024 * 1024),
ctl_out_w,
));
let mut ctl_out = FrameReader::new(ctl_out_test_r, 1024 * 1024);
match ctl_out.next().await.unwrap() {
Some(Inbound::Frame(f)) => {
assert_eq!(f["id"], 1, "the echoed initialize answers our id: {f}");
assert_eq!(
f["params"]["_meta"]["mcpmesh/service"], "echo",
"the peer saw the service-injected initialize (the one enumerated \
edit), echoed verbatim: {f}"
);
}
other => panic!("the echo must reach the control side, got {other:?}"),
}
assert!(
ctl_out.next().await.unwrap().is_none(),
"the peer closing ends the session cleanly (control-side EOF)"
);
session.await.unwrap().expect("pipe_session returns Ok");
let _ = done_tx.send(()); assert_eq!(
peer.await.unwrap(),
vec![inject_service(init, "echo")],
"the peer received exactly the injected initialize before the half-close"
);
})
.await
.expect("pipe_session drain test timed out");
}
#[tokio::test(flavor = "multi_thread")]
async fn a_raced_dial_carries_the_stored_hints() {
let dir = tempfile::tempdir().unwrap();
let cfg = dir.path().join("config.toml");
std::fs::write(&cfg, "").unwrap();
let mesh = crate::daemon::testutil::hermetic_mesh(cfg).await;
let first = *iroh::SecretKey::from_bytes(&[61u8; 32]).public().as_bytes();
let second = *iroh::SecretKey::from_bytes(&[62u8; 32]).public().as_bytes();
let sock: std::net::SocketAddr = "127.0.0.1:4455".parse().unwrap();
let hint = serde_json::to_string(&iroh::EndpointAddr::from_parts(
iroh::EndpointId::from_bytes(&first).unwrap(),
[iroh::TransportAddr::Ip(sock)],
))
.unwrap();
for (eid, nick, last) in [
(first, "alice-laptop", Some(hint.clone())),
(second, "alice-phone", None),
] {
mesh.store
.add(crate::allowlist::PeerEntry {
endpoint_id: eid,
nickname: nick.into(),
services: vec![],
paired_at: None,
user_id: Some("b64u:alice".into()),
last_addr: last,
})
.unwrap();
}
let addrs = super::hinted_addrs(&mesh, vec![first, second])
.await
.expect("candidates resolve");
assert_eq!(addrs.len(), 2, "every candidate stays a candidate");
assert!(
!addrs[0].addrs.is_empty(),
"a raced candidate must carry its stored hint — a bare id is discovery-only, which is \
exactly what made a two-device person unreachable on a LAN with no discovery"
);
assert_eq!(
*addrs[0].id.as_bytes(),
first,
"and it must be THAT device's address"
);
assert!(addrs[1].addrs.is_empty());
assert_eq!(*addrs[1].id.as_bytes(), second);
}
#[test]
fn stored_dial_addr_attaches_validates_and_degrades() {
let id = iroh::SecretKey::from_bytes(&[7u8; 32]).public();
let other = iroh::SecretKey::from_bytes(&[8u8; 32]).public();
let sock: std::net::SocketAddr = "127.0.0.1:4444".parse().unwrap();
let stored = iroh::EndpointAddr::from_parts(id, [iroh::TransportAddr::Ip(sock)]);
let stored_json = serde_json::to_string(&stored).unwrap();
let addr = stored_dial_addr(Some(&stored_json), id);
assert_eq!(addr, stored, "a matching-id hint is dialed as stored");
assert_eq!(stored_dial_addr(None, id), iroh::EndpointAddr::from(id));
assert_eq!(
stored_dial_addr(Some("not json"), id),
iroh::EndpointAddr::from(id)
);
let mismatched = serde_json::to_string(&iroh::EndpointAddr::from_parts(
other,
[iroh::TransportAddr::Ip(sock)],
))
.unwrap();
assert_eq!(
stored_dial_addr(Some(&mismatched), id),
iroh::EndpointAddr::from(id)
);
}
#[tokio::test]
async fn connect_with_timeout_fails_fast_on_an_unreachable_peer() {
let ep = iroh::Endpoint::builder(iroh::endpoint::presets::Minimal)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![mcpmesh_net::ALPN_MCP.to_vec()])
.bind()
.await
.unwrap();
let dead = iroh::EndpointAddr::from(iroh::EndpointId::from_bytes(&[3u8; 32]).unwrap());
let start = std::time::Instant::now();
let r = super::connect_with_timeout(
&ep,
dead,
"svc",
std::time::Duration::from_millis(300),
None,
)
.await;
assert!(r.is_err(), "an unreachable dial times out to Err");
assert!(
start.elapsed() < std::time::Duration::from_secs(3),
"the explicit timeout fired fast"
);
}
#[test]
fn the_per_session_config_carries_this_nodes_keepalive_not_irohs_default() {
let cfg = super::per_session_transport_config(Some(3), 2)
.expect("3s is legal against a 2s keepalive")
.expect("a value was supplied, so a config is built");
let d = format!("{cfg:?}");
assert!(
d.contains("max_idle_timeout: Some(3000)"),
"the caller's idle timeout must be set: {d}"
);
assert!(
d.contains("keep_alive_interval: Some(2s)"),
"this NODE's keepalive must survive — a fresh builder resets it to iroh's 5s, which \
would make the 3s idle timeout sever a healthy session: {d}"
);
assert!(
d.contains("default_path_keep_alive_interval: Some(2s)"),
"…including the PATH keepalive, which is the one iroh actually pings on: {d}"
);
let cfg = super::per_session_transport_config(Some(0), 2)
.unwrap()
.unwrap();
let d = format!("{cfg:?}");
assert!(d.contains("max_idle_timeout: None"), "{d}");
assert!(d.contains("keep_alive_interval: Some(2s)"), "{d}");
}
#[test]
fn a_per_session_idle_timeout_resolves_or_refuses_but_never_silently_ignores() {
assert!(
super::per_session_transport_config(None, 5)
.unwrap()
.is_none()
);
assert!(
super::per_session_transport_config(Some(30), 5)
.unwrap()
.is_some()
);
for bad in [1u64, 4, 5] {
let e = super::per_session_transport_config(Some(bad), 5)
.expect_err("at or below the keepalive must be refused");
let msg = format!("{e:#}");
assert!(
msg.contains("keepalive") && msg.contains(&bad.to_string()),
"the refusal must name the value and the reason: {msg}"
);
}
assert!(super::per_session_transport_config(Some(3), 2).is_ok());
assert!(super::per_session_transport_config(Some(2), 2).is_err());
let e = super::per_session_transport_config(Some(u64::MAX), 5)
.expect_err("an unencodable value must be refused");
assert!(
format!("{e:#}").contains("out of the range QUIC can encode"),
"{e:#}"
);
}
}