mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
fix(desktop): keep mesh allowlist on transient roster query failure (#2024)
This commit is contained in:
@@ -64,6 +64,17 @@ async fn query_mesh_discovery_events(state: &AppState) -> Result<Vec<nostr::Even
|
||||
let mut events = relay::query_relay(state, &[mesh_llm::relay_membership_filter()]).await?;
|
||||
let member_pubkeys = mesh_llm::current_member_pubkeys(&events);
|
||||
if member_pubkeys.is_empty() {
|
||||
// Distinguish "relay returned a membership snapshot listing zero
|
||||
// members" (authoritative empty — allowed to shrink the roster to
|
||||
// self-only) from "no membership snapshot came back at all" (a
|
||||
// transient gap / replication lag). The relay publishes an explicit
|
||||
// kind:13534 event even for a zero-member community, so its absence
|
||||
// means the query is incomplete: surface it as an error so the
|
||||
// reconcile loop keeps the current allowlist instead of flapping the
|
||||
// node down to self-only on a successful-but-empty response.
|
||||
if !mesh_llm::has_membership_snapshot(&events) {
|
||||
return Err("relay returned no membership snapshot".to_string());
|
||||
}
|
||||
return Ok(events);
|
||||
}
|
||||
let mut status_filter = mesh_llm::mesh_status_filter();
|
||||
@@ -88,12 +99,25 @@ async fn query_mesh_discovery_events(state: &AppState) -> Result<Vec<nostr::Even
|
||||
}
|
||||
|
||||
/// Resolve the admission roster by intersecting member-signed mesh status
|
||||
/// reporters with the current NIP-43 direct-member list. Missing membership or
|
||||
/// a failed query returns an empty roster, which the runtime normalizes to
|
||||
/// self-only admission.
|
||||
pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Vec<String> {
|
||||
match query_mesh_discovery_events(state).await {
|
||||
Ok(events) => mesh_llm::owner_ids_from_events(&events),
|
||||
/// reporters with the current NIP-43 direct-member list.
|
||||
///
|
||||
/// Returns `Err` when the relay query fails. Callers MUST distinguish this from
|
||||
/// an `Ok(empty)` roster (a genuinely empty community): a failed query must
|
||||
/// never be collapsed into "self-only", or a transient relay blip de-admits
|
||||
/// every other member. `reconcile_roster` relies on this to keep the current
|
||||
/// allowlist on error instead of restarting the node down to self-only.
|
||||
pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Result<Vec<String>, String> {
|
||||
let events = query_mesh_discovery_events(state).await?;
|
||||
Ok(mesh_llm::owner_ids_from_events(&events))
|
||||
}
|
||||
|
||||
/// Resolve the roster for an initial node *start*, failing closed to self-only
|
||||
/// (an empty roster) when the relay query fails. This is safe only at start:
|
||||
/// there is no established allowlist to preserve yet. The periodic
|
||||
/// `reconcile_roster` path must NOT use this — it has a live roster to keep.
|
||||
pub(crate) async fn resolve_trusted_owner_ids_or_self_only(state: &AppState) -> Vec<String> {
|
||||
match resolve_trusted_owner_ids(state).await {
|
||||
Ok(owners) => owners,
|
||||
Err(error) => {
|
||||
eprintln!("buzz-mesh: roster query failed; allowing only this node: {error}");
|
||||
Vec::new()
|
||||
@@ -117,7 +141,7 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C
|
||||
model_id: Some(config.model_id),
|
||||
max_vram_gb: config.max_vram_gb,
|
||||
join_token: None,
|
||||
trusted_owner_ids: Some(resolve_trusted_owner_ids(state).await),
|
||||
trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await),
|
||||
};
|
||||
let started = mesh_llm::DesktopMeshRuntime::start(request)
|
||||
.await
|
||||
@@ -137,7 +161,7 @@ pub async fn mesh_start_node(
|
||||
// Frontend requests never carry a roster; resolve it here so every
|
||||
// UI-started node enforces the member allowlist.
|
||||
if request.trusted_owner_ids.is_none() {
|
||||
request.trusted_owner_ids = Some(resolve_trusted_owner_ids(&state).await);
|
||||
request.trusted_owner_ids = Some(resolve_trusted_owner_ids_or_self_only(&state).await);
|
||||
}
|
||||
let mut runtime = state.mesh_llm_runtime.lock().await;
|
||||
if runtime.is_some() {
|
||||
@@ -268,7 +292,7 @@ pub(crate) async fn ensure_client_node_for_model(
|
||||
model_id: None,
|
||||
max_vram_gb: None,
|
||||
join_token: Some(join_token),
|
||||
trusted_owner_ids: Some(resolve_trusted_owner_ids(state).await),
|
||||
trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await),
|
||||
};
|
||||
let mut runtime = state.mesh_llm_runtime.lock().await;
|
||||
if runtime.is_some() {
|
||||
|
||||
@@ -51,10 +51,13 @@ pub async fn start_coordinator(app: AppHandle) {
|
||||
});
|
||||
let roster_app = app.clone();
|
||||
let roster_watcher = tokio::spawn(async move {
|
||||
// Carries a shrink awaiting confirmation across polls (hysteresis):
|
||||
// a reduced roster must be seen twice in a row before we tear down.
|
||||
let mut pending_shrink: Option<Vec<String>> = None;
|
||||
loop {
|
||||
tokio::time::sleep(ROSTER_POLL_INTERVAL).await;
|
||||
let state = roster_app.state::<AppState>();
|
||||
if let Err(error) = reconcile_roster(&state).await {
|
||||
if let Err(error) = reconcile_roster(&state, &mut pending_shrink).await {
|
||||
eprintln!("buzz-mesh: roster reconcile failed: {error}");
|
||||
}
|
||||
}
|
||||
@@ -73,21 +76,108 @@ pub async fn start_coordinator(app: AppHandle) {
|
||||
}
|
||||
}
|
||||
|
||||
async fn reconcile_roster(state: &AppState) -> Result<(), String> {
|
||||
/// Outcome of a roster reconcile decision.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum RosterReconcileAction {
|
||||
/// Keep the running allowlist untouched (no-op, or a failure we ride out).
|
||||
Keep,
|
||||
/// Restart the node with a freshly resolved roster.
|
||||
Restart(Vec<String>),
|
||||
/// Observed a *shrink* (or empty) once. Hold the current allowlist and
|
||||
/// require the same reduced roster on the next poll before tearing down,
|
||||
/// so a single transient short-read never drops a member mid-inference.
|
||||
AwaitConfirm(Vec<String>),
|
||||
}
|
||||
|
||||
/// Whether `fresh` removes any owner present in `current` (a shrink), as
|
||||
/// opposed to purely adding owners or leaving the set unchanged.
|
||||
fn roster_shrinks(current: &[String], fresh: &[String]) -> bool {
|
||||
current.iter().any(|owner| !fresh.contains(owner))
|
||||
}
|
||||
|
||||
/// Pure decision for `reconcile_roster`, extracted so the transient-failure
|
||||
/// and hysteresis invariants are unit-testable without a live relay.
|
||||
///
|
||||
/// `pending_shrink` is the reduced roster we are waiting to re-confirm (from a
|
||||
/// prior poll's [`RosterReconcileAction::AwaitConfirm`]), if any.
|
||||
///
|
||||
/// Rules:
|
||||
/// - query failed (`Err`) → `Keep` (never de-admit on a relay blip)
|
||||
/// - resolved roster == current → `Keep` (no-op)
|
||||
/// - grows (only additions) → `Restart` immediately (fast admission)
|
||||
/// - shrinks/empties, first observation → `AwaitConfirm` (hold, re-check next poll)
|
||||
/// - shrinks/empties, confirmed → `Restart` (same reduced roster twice)
|
||||
fn roster_reconcile_action(
|
||||
current_owners: &[String],
|
||||
pending_shrink: Option<&[String]>,
|
||||
query: Result<Vec<String>, String>,
|
||||
) -> RosterReconcileAction {
|
||||
let fresh = match query {
|
||||
Err(error) => {
|
||||
eprintln!(
|
||||
"buzz-mesh: roster reconcile query failed; keeping current allowlist: {error}"
|
||||
);
|
||||
return RosterReconcileAction::Keep;
|
||||
}
|
||||
Ok(fresh) => fresh,
|
||||
};
|
||||
|
||||
if fresh == current_owners {
|
||||
return RosterReconcileAction::Keep;
|
||||
}
|
||||
|
||||
// Growth (pure additions) is safe to apply immediately.
|
||||
if !roster_shrinks(current_owners, &fresh) {
|
||||
return RosterReconcileAction::Restart(fresh);
|
||||
}
|
||||
|
||||
// A shrink (including down to empty) must be confirmed across two
|
||||
// consecutive polls with the *same* reduced roster before we tear down.
|
||||
match pending_shrink {
|
||||
Some(pending) if pending == fresh => RosterReconcileAction::Restart(fresh),
|
||||
_ => RosterReconcileAction::AwaitConfirm(fresh),
|
||||
}
|
||||
}
|
||||
|
||||
async fn reconcile_roster(
|
||||
state: &AppState,
|
||||
pending_shrink: &mut Option<Vec<String>>,
|
||||
) -> Result<(), String> {
|
||||
let current_request = {
|
||||
let runtime = state.mesh_llm_runtime.lock().await;
|
||||
match runtime.as_ref() {
|
||||
Some(runtime) => runtime.start_request().clone(),
|
||||
None => return Ok(()),
|
||||
None => {
|
||||
*pending_shrink = None;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
};
|
||||
let Some(current_owners) = current_request.trusted_owner_ids.as_ref() else {
|
||||
*pending_shrink = None;
|
||||
return Ok(());
|
||||
};
|
||||
let fresh = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await;
|
||||
if &fresh == current_owners {
|
||||
return Ok(());
|
||||
}
|
||||
// A failed roster query must NOT be treated as "the roster became empty":
|
||||
// doing so would restart the node down to self-only and de-admit every
|
||||
// other member on a transient relay blip (the flapping restart loop). Keep
|
||||
// the current allowlist and try again on the next poll. A shrink is held
|
||||
// for one extra poll (hysteresis) so a single short-read never tears down.
|
||||
let query = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await;
|
||||
let fresh = match roster_reconcile_action(current_owners, pending_shrink.as_deref(), query) {
|
||||
RosterReconcileAction::Keep => {
|
||||
*pending_shrink = None;
|
||||
return Ok(());
|
||||
}
|
||||
RosterReconcileAction::AwaitConfirm(reduced) => {
|
||||
eprintln!("buzz-mesh: roster shrink observed; awaiting confirmation before restart");
|
||||
*pending_shrink = Some(reduced);
|
||||
return Ok(());
|
||||
}
|
||||
RosterReconcileAction::Restart(fresh) => {
|
||||
*pending_shrink = None;
|
||||
fresh
|
||||
}
|
||||
};
|
||||
|
||||
let mut request = current_request;
|
||||
request.trusted_owner_ids = Some(fresh);
|
||||
@@ -227,6 +317,91 @@ mod tests {
|
||||
|
||||
use super::*;
|
||||
|
||||
// Regression: a transient roster-query failure must never restart the node
|
||||
// down to self-only. Before the fix, `resolve_trusted_owner_ids` returned
|
||||
// an empty Vec on error, which `reconcile_roster` read as "roster changed
|
||||
// to empty" and restarted — de-admitting every other member and flapping
|
||||
// the node on each relay blip. See #2000 follow-up.
|
||||
#[test]
|
||||
fn failed_roster_query_keeps_current_allowlist() {
|
||||
let current = vec!["owner-a".to_string(), "owner-b".to_string()];
|
||||
let action = roster_reconcile_action(¤t, None, Err("relay returned 503".to_string()));
|
||||
assert_eq!(
|
||||
action,
|
||||
RosterReconcileAction::Keep,
|
||||
"a failed query must keep the running allowlist, never de-admit members"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unchanged_roster_is_a_noop() {
|
||||
let current = vec!["owner-a".to_string()];
|
||||
let action = roster_reconcile_action(¤t, None, Ok(vec!["owner-a".to_string()]));
|
||||
assert_eq!(action, RosterReconcileAction::Keep);
|
||||
}
|
||||
|
||||
// Growth (pure additions) applies immediately — fast admission is fine.
|
||||
#[test]
|
||||
fn roster_growth_restarts_immediately() {
|
||||
let current = vec!["owner-a".to_string()];
|
||||
let fresh = vec!["owner-a".to_string(), "owner-c".to_string()];
|
||||
let action = roster_reconcile_action(¤t, None, Ok(fresh.clone()));
|
||||
assert_eq!(action, RosterReconcileAction::Restart(fresh));
|
||||
}
|
||||
|
||||
// A shrink is NOT applied on first observation — it must be confirmed.
|
||||
#[test]
|
||||
fn roster_shrink_awaits_confirmation_first() {
|
||||
let current = vec!["owner-a".to_string(), "owner-b".to_string()];
|
||||
let reduced = vec!["owner-a".to_string()];
|
||||
let action = roster_reconcile_action(¤t, None, Ok(reduced.clone()));
|
||||
assert_eq!(action, RosterReconcileAction::AwaitConfirm(reduced));
|
||||
}
|
||||
|
||||
// The same reduced roster on two consecutive polls confirms the shrink.
|
||||
#[test]
|
||||
fn roster_shrink_restarts_once_confirmed() {
|
||||
let current = vec!["owner-a".to_string(), "owner-b".to_string()];
|
||||
let reduced = vec!["owner-a".to_string()];
|
||||
let action = roster_reconcile_action(¤t, Some(&reduced), Ok(reduced.clone()));
|
||||
assert_eq!(action, RosterReconcileAction::Restart(reduced));
|
||||
}
|
||||
|
||||
// A shrink that changes between polls is not confirmed — it re-holds with
|
||||
// the newly observed reduced roster instead of tearing down.
|
||||
#[test]
|
||||
fn roster_shrink_reconfirms_when_it_changes() {
|
||||
let current = vec!["a".to_string(), "b".to_string(), "c".to_string()];
|
||||
let first_reduced = vec!["a".to_string(), "b".to_string()];
|
||||
let second_reduced = vec!["a".to_string()];
|
||||
let action =
|
||||
roster_reconcile_action(¤t, Some(&first_reduced), Ok(second_reduced.clone()));
|
||||
assert_eq!(action, RosterReconcileAction::AwaitConfirm(second_reduced));
|
||||
}
|
||||
|
||||
// A genuinely empty community (Ok(empty), distinct from a failed query)
|
||||
// still shrinks to self-only — but only after confirmation.
|
||||
#[test]
|
||||
fn genuinely_empty_roster_awaits_then_restarts_to_self_only() {
|
||||
let current = vec!["owner-a".to_string()];
|
||||
let first = roster_reconcile_action(¤t, None, Ok(Vec::new()));
|
||||
assert_eq!(first, RosterReconcileAction::AwaitConfirm(Vec::new()));
|
||||
let empty: Vec<String> = Vec::new();
|
||||
let confirmed = roster_reconcile_action(¤t, Some(&empty), Ok(Vec::new()));
|
||||
assert_eq!(confirmed, RosterReconcileAction::Restart(Vec::new()));
|
||||
}
|
||||
|
||||
// A shrink followed by recovery to the full roster cancels the teardown.
|
||||
#[test]
|
||||
fn roster_shrink_then_recovery_keeps_allowlist() {
|
||||
let current = vec!["owner-a".to_string(), "owner-b".to_string()];
|
||||
let reduced = vec!["owner-a".to_string()];
|
||||
let held = roster_reconcile_action(¤t, None, Ok(reduced.clone()));
|
||||
assert_eq!(held, RosterReconcileAction::AwaitConfirm(reduced.clone()));
|
||||
let recovered = roster_reconcile_action(¤t, Some(&reduced), Ok(current.clone()));
|
||||
assert_eq!(recovered, RosterReconcileAction::Keep);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn member_heartbeat_leaves_room_before_admission_status_expires() {
|
||||
assert!(
|
||||
|
||||
@@ -93,6 +93,20 @@ pub(crate) fn current_member_pubkeys(events: &[nostr::Event]) -> Vec<String> {
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Whether the relay actually returned a NIP-43 membership snapshot (kind
|
||||
/// 13534) in `events`.
|
||||
///
|
||||
/// The relay publishes an explicit membership event even for a zero-member
|
||||
/// community, so its presence is what makes an empty roster *authoritative*.
|
||||
/// Callers use this to distinguish "the community genuinely has no members"
|
||||
/// (snapshot present, zero `member` tags) from "no snapshot came back at all"
|
||||
/// (a transient relay gap / replication lag). Only the former may shrink the
|
||||
/// admission roster; the latter must be surfaced as an error so the reconcile
|
||||
/// loop keeps the current allowlist instead of restarting to self-only.
|
||||
pub(crate) fn has_membership_snapshot(events: &[nostr::Event]) -> bool {
|
||||
events.iter().any(|event| event.kind.as_u16() == 13_534)
|
||||
}
|
||||
|
||||
fn owner_id_from_status_event(event: &nostr::Event) -> Option<String> {
|
||||
let content = serde_json::from_str::<serde_json::Value>(&event.content).ok()?;
|
||||
let owner_id = content
|
||||
@@ -200,11 +214,12 @@ pub fn availability_from_events(events: Vec<nostr::Event>) -> MeshAvailability {
|
||||
.unwrap_or_default()
|
||||
.into_iter()
|
||||
.filter_map(|mut target| {
|
||||
let endpoint_id =
|
||||
let validated =
|
||||
super::transport_policy::validate_advertised_endpoint(&target.endpoint_addr)
|
||||
.ok()?;
|
||||
target.endpoint_addr = validated.join_token;
|
||||
if target.endpoint_id.is_none() {
|
||||
target.endpoint_id = Some(endpoint_id);
|
||||
target.endpoint_id = Some(validated.endpoint_id);
|
||||
}
|
||||
if target.device_id.is_none() {
|
||||
target.device_id = target.endpoint_id.clone();
|
||||
@@ -322,7 +337,9 @@ pub(super) fn device_name_from_status(
|
||||
}
|
||||
|
||||
fn endpoint_id_from_invite_token(invite_token: &str) -> Option<String> {
|
||||
super::transport_policy::validate_advertised_endpoint(invite_token).ok()
|
||||
super::transport_policy::validate_advertised_endpoint(invite_token)
|
||||
.ok()
|
||||
.map(|validated| validated.endpoint_id)
|
||||
}
|
||||
|
||||
fn string_value(value: &serde_json::Value, key: &str) -> Option<String> {
|
||||
|
||||
@@ -8,7 +8,9 @@ mod discovery;
|
||||
pub use discovery::{
|
||||
availability_from_events, mesh_status_filter, owner_ids_from_events, relay_membership_filter,
|
||||
};
|
||||
pub(crate) use discovery::{current_member_pubkeys, MESH_STATUS_PAGE_SIZE};
|
||||
pub(crate) use discovery::{
|
||||
current_member_pubkeys, has_membership_snapshot, MESH_STATUS_PAGE_SIZE,
|
||||
};
|
||||
use discovery::{device_name_from_status, endpoint_id_from_status, enrich_status_payload_identity};
|
||||
|
||||
mod catalog;
|
||||
@@ -270,8 +272,8 @@ async fn ensure_model_downloaded(model: &str) -> anyhow::Result<()> {
|
||||
}
|
||||
|
||||
impl DesktopMeshRuntime {
|
||||
pub async fn start(request: StartMeshNodeRequest) -> anyhow::Result<Self> {
|
||||
validate_no_leak_request(&request)?;
|
||||
pub async fn start(mut request: StartMeshNodeRequest) -> anyhow::Result<Self> {
|
||||
sanitize_no_leak_request(&mut request)?;
|
||||
initialize_mesh_native_runtime().await?;
|
||||
let model_id = request
|
||||
.model_id
|
||||
@@ -420,8 +422,8 @@ impl DesktopMeshRuntime {
|
||||
|
||||
pub async fn dial_endpoint_addr(&self, endpoint_addr: impl Into<String>) -> anyhow::Result<()> {
|
||||
let endpoint_addr = endpoint_addr.into();
|
||||
validate_advertised_endpoint(&endpoint_addr)?;
|
||||
self.handle.join_token(endpoint_addr).await
|
||||
let validated = validate_advertised_endpoint(&endpoint_addr)?;
|
||||
self.handle.join_token(validated.join_token).await
|
||||
}
|
||||
|
||||
pub async fn installed_models(&self) -> anyhow::Result<Vec<MeshModelOption>> {
|
||||
@@ -503,9 +505,9 @@ fn normalized_roster(
|
||||
Some(owners)
|
||||
}
|
||||
|
||||
fn validate_no_leak_request(request: &StartMeshNodeRequest) -> anyhow::Result<()> {
|
||||
if let Some(join_token) = request.join_token.as_deref() {
|
||||
validate_advertised_endpoint(join_token)?;
|
||||
fn sanitize_no_leak_request(request: &mut StartMeshNodeRequest) -> anyhow::Result<()> {
|
||||
if let Some(join_token) = request.join_token.as_mut() {
|
||||
*join_token = validate_advertised_endpoint(join_token)?.join_token;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -214,6 +214,34 @@ fn signed_membership_event(members: &[String]) -> nostr::Event {
|
||||
.expect("test membership event signs")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn has_membership_snapshot_distinguishes_empty_from_missing() {
|
||||
// A zero-member community still publishes an explicit kind:13534 event, so
|
||||
// its presence — not the member count — is what makes an empty roster
|
||||
// authoritative. No snapshot at all means the query was incomplete.
|
||||
let zero_member_snapshot = signed_membership_event(&[]);
|
||||
assert!(
|
||||
super::has_membership_snapshot(std::slice::from_ref(&zero_member_snapshot)),
|
||||
"an explicit zero-member snapshot counts as present"
|
||||
);
|
||||
|
||||
let member = nostr::Keys::parse(&"1".repeat(64))
|
||||
.unwrap()
|
||||
.public_key()
|
||||
.to_hex();
|
||||
let populated = signed_membership_event(std::slice::from_ref(&member));
|
||||
assert!(super::has_membership_snapshot(std::slice::from_ref(
|
||||
&populated
|
||||
)));
|
||||
|
||||
// A response with only status events (or nothing) has no snapshot.
|
||||
assert!(!super::has_membership_snapshot(&[]));
|
||||
let status_only = signed_reporter_status(&"2".repeat(64), "owner-x");
|
||||
assert!(!super::has_membership_snapshot(std::slice::from_ref(
|
||||
&status_only
|
||||
)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn owner_ids_from_events_collects_sorted_deduped_roster() {
|
||||
let secret_a = "1".repeat(64);
|
||||
|
||||
@@ -72,7 +72,15 @@ fn parse_configured_relay_url(raw: &str) -> anyhow::Result<RelayUrl> {
|
||||
.map_err(|error| anyhow::anyhow!("invalid iroh relay URL {raw:?}: {error}"))
|
||||
}
|
||||
|
||||
pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result<String> {
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(super) struct ValidatedEndpoint {
|
||||
pub endpoint_id: String,
|
||||
pub join_token: String,
|
||||
}
|
||||
|
||||
pub(super) fn validate_advertised_endpoint(
|
||||
invite_token: &str,
|
||||
) -> anyhow::Result<ValidatedEndpoint> {
|
||||
let mode = iroh_relay_mode()?;
|
||||
validate_advertised_endpoint_with_mode(invite_token, &mode)
|
||||
}
|
||||
@@ -80,7 +88,7 @@ pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result
|
||||
pub(super) fn validate_advertised_endpoint_with_mode(
|
||||
invite_token: &str,
|
||||
mode: &IrohRelayMode,
|
||||
) -> anyhow::Result<String> {
|
||||
) -> anyhow::Result<ValidatedEndpoint> {
|
||||
let token = invite_token.trim();
|
||||
if token.is_empty() {
|
||||
anyhow::bail!("mesh invite token is empty");
|
||||
@@ -92,59 +100,139 @@ pub(super) fn validate_advertised_endpoint_with_mode(
|
||||
.decode(token)
|
||||
.map_err(|error| anyhow::anyhow!("invalid mesh invite encoding: {error}"))?;
|
||||
|
||||
let addrs = if let Ok(addr) = serde_json::from_slice::<EndpointAddr>(&payload) {
|
||||
vec![addr]
|
||||
} else {
|
||||
let signed = serde_json::from_slice::<SignedBootstrapToken>(&payload)
|
||||
.map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?;
|
||||
signed
|
||||
.verify()
|
||||
.map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?;
|
||||
if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS
|
||||
{
|
||||
anyhow::bail!(
|
||||
"signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses"
|
||||
);
|
||||
}
|
||||
signed
|
||||
.serialized_addrs
|
||||
.iter()
|
||||
.map(|bytes| {
|
||||
serde_json::from_slice::<EndpointAddr>(bytes)
|
||||
.map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}"))
|
||||
})
|
||||
.collect::<anyhow::Result<Vec<_>>>()?
|
||||
};
|
||||
|
||||
for addr in &addrs {
|
||||
validate_endpoint_addr(addr, mode)?;
|
||||
if let Ok(mut addr) = serde_json::from_slice::<EndpointAddr>(&payload) {
|
||||
retain_usable_transports(&mut addr, mode)?;
|
||||
let endpoint_id = addr.id.to_string();
|
||||
let join_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&addr)?);
|
||||
return Ok(ValidatedEndpoint {
|
||||
endpoint_id,
|
||||
join_token,
|
||||
});
|
||||
}
|
||||
Ok(addrs[0].id.to_string())
|
||||
|
||||
let signed = serde_json::from_slice::<SignedBootstrapToken>(&payload)
|
||||
.map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?;
|
||||
signed
|
||||
.verify()
|
||||
.map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?;
|
||||
if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS {
|
||||
anyhow::bail!(
|
||||
"signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses"
|
||||
);
|
||||
}
|
||||
let addrs = signed
|
||||
.serialized_addrs
|
||||
.iter()
|
||||
.map(|bytes| {
|
||||
serde_json::from_slice::<EndpointAddr>(bytes)
|
||||
.map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}"))
|
||||
})
|
||||
.collect::<anyhow::Result<Vec<_>>>()?;
|
||||
|
||||
// Rewriting a signed token would invalidate its signature. Keep the signed
|
||||
// envelope intact only when every advertised transport is policy-approved;
|
||||
// mixed signed tokens fail closed rather than leaking rejected dial targets.
|
||||
for addr in &addrs {
|
||||
validate_signed_transports(addr, mode)?;
|
||||
}
|
||||
Ok(ValidatedEndpoint {
|
||||
endpoint_id: addrs[0].id.to_string(),
|
||||
join_token: token.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_endpoint_addr(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> {
|
||||
fn retain_usable_transports(addr: &mut EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> {
|
||||
validate_transport_count(addr)?;
|
||||
let mut rejections = Vec::new();
|
||||
addr.addrs
|
||||
.retain(|transport| match validate_transport(transport, mode) {
|
||||
Ok(()) => true,
|
||||
Err(error) => {
|
||||
rejections.push(error.to_string());
|
||||
false
|
||||
}
|
||||
});
|
||||
if addr.addrs.is_empty() {
|
||||
anyhow::bail!(
|
||||
"mesh endpoint has no usable transport address (all rejected: {})",
|
||||
rejections.join("; ")
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_signed_transports(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> {
|
||||
validate_transport_count(addr)?;
|
||||
let mut usable = 0usize;
|
||||
for transport in &addr.addrs {
|
||||
match validate_transport(transport, mode) {
|
||||
Ok(()) => usable += 1,
|
||||
// mesh-llm currently signs its own port-0 placeholder alongside a
|
||||
// valid relay. It is non-dialable, so preserving it is safe and
|
||||
// necessary for stock signed tokens to remain usable.
|
||||
Err(_) if matches!(transport, TransportAddr::Ip(socket) if socket.port() == 0) => {}
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
}
|
||||
if usable == 0 {
|
||||
anyhow::bail!("signed mesh endpoint has no usable transport address");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_transport_count(addr: &EndpointAddr) -> anyhow::Result<()> {
|
||||
if addr.addrs.is_empty() || addr.addrs.len() > MAX_ENDPOINT_TRANSPORT_ADDRS {
|
||||
anyhow::bail!(
|
||||
"mesh endpoint must contain 1..={MAX_ENDPOINT_TRANSPORT_ADDRS} transport addresses"
|
||||
);
|
||||
}
|
||||
for transport in &addr.addrs {
|
||||
match transport {
|
||||
TransportAddr::Relay(relay) if relay_allowed(relay, mode) => {}
|
||||
TransportAddr::Relay(relay) => {
|
||||
anyhow::bail!("mesh endpoint advertises unapproved relay URL {relay}")
|
||||
}
|
||||
TransportAddr::Ip(socket) => validate_direct_socket(*socket)?,
|
||||
_ => anyhow::bail!("mesh endpoint contains an unsupported transport address"),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_transport(transport: &TransportAddr, mode: &IrohRelayMode) -> anyhow::Result<()> {
|
||||
match transport {
|
||||
TransportAddr::Relay(relay) if relay_allowed(relay, mode) => Ok(()),
|
||||
TransportAddr::Relay(relay) => anyhow::bail!("unapproved relay URL {relay}"),
|
||||
TransportAddr::Ip(socket) => validate_direct_socket(*socket),
|
||||
_ => anyhow::bail!("unsupported transport address"),
|
||||
}
|
||||
}
|
||||
|
||||
/// mesh-llm's default public relay set (`RelayPolicy::DefaultPublic` in
|
||||
/// `mesh-llm-host-runtime`). A stock mesh-llm server with no custom relay
|
||||
/// config advertises endpoints on exactly these relays, so buzz's `Default`
|
||||
/// mode MUST accept them — otherwise shared compute rejects every out-of-the-box
|
||||
/// mesh-llm serving node (they are not in iroh's own prod relay map).
|
||||
///
|
||||
/// Kept in sync with `effective_relay_urls(RelayPolicy::DefaultPublic, &[])`.
|
||||
const MESH_LLM_DEFAULT_RELAYS: &[&str] = &[
|
||||
"https://usw1-2.relay.michaelneale.mesh-llm.iroh.link./",
|
||||
"https://aps1-1.relay.michaelneale.mesh-llm.iroh.link./",
|
||||
];
|
||||
|
||||
/// Whether `relay` is one of mesh-llm's baked-in default public relays.
|
||||
/// Parses each known URL to a `RelayUrl` so comparison is normalization-safe
|
||||
/// (matches regardless of trailing-dot / trailing-slash formatting).
|
||||
fn is_mesh_llm_default_relay(relay: &RelayUrl) -> bool {
|
||||
MESH_LLM_DEFAULT_RELAYS.iter().any(|candidate| {
|
||||
candidate
|
||||
.parse::<RelayUrl>()
|
||||
.map(|known| &known == relay)
|
||||
.unwrap_or(false)
|
||||
})
|
||||
}
|
||||
|
||||
fn relay_allowed(relay: &RelayUrl, mode: &IrohRelayMode) -> bool {
|
||||
match mode {
|
||||
IrohRelayMode::Disabled => false,
|
||||
IrohRelayMode::Default => iroh::defaults::prod::default_relay_map().contains(relay),
|
||||
// `Default` covers both iroh's own production relays AND mesh-llm's
|
||||
// default public relays. Without the latter, a stock mesh-llm serving
|
||||
// node is unreachable by default and shared compute silently fails with
|
||||
// "no live member is serving this model" even though discovery found it.
|
||||
IrohRelayMode::Default => {
|
||||
iroh::defaults::prod::default_relay_map().contains(relay)
|
||||
|| is_mesh_llm_default_relay(relay)
|
||||
}
|
||||
IrohRelayMode::Custom(urls) => urls.contains(relay),
|
||||
}
|
||||
}
|
||||
@@ -213,6 +301,76 @@ mod tests {
|
||||
assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Disabled).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn default_mode_accepts_meshllm_default_relays() {
|
||||
// Regression: a stock mesh-llm serving node advertises endpoints on
|
||||
// mesh-llm's OWN default public relays (not iroh's prod relay map).
|
||||
// Under `Default` mode these MUST be accepted, or shared compute rejects
|
||||
// every out-of-the-box mesh-llm server with "no live member is serving
|
||||
// this model" even though discovery found it. See mesh-llm
|
||||
// effective_relay_urls(RelayPolicy::DefaultPublic, &[]).
|
||||
for relay_url in MESH_LLM_DEFAULT_RELAYS {
|
||||
let relay: RelayUrl = relay_url
|
||||
.parse()
|
||||
.unwrap_or_else(|e| panic!("mesh-llm default relay {relay_url:?} must parse: {e}"));
|
||||
assert!(
|
||||
relay_allowed(&relay, &IrohRelayMode::Default),
|
||||
"Default mode must accept mesh-llm default relay {relay_url}"
|
||||
);
|
||||
|
||||
// And end-to-end through the advertised-endpoint validator.
|
||||
let token = endpoint_token_for_test([TransportAddr::Relay(relay)]);
|
||||
assert!(
|
||||
validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_ok(),
|
||||
"Default mode must validate an endpoint on mesh-llm default relay {relay_url}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_with_one_good_and_one_junk_candidate_is_sanitized() {
|
||||
// A mesh-llm endpoint can advertise a usable relay alongside an
|
||||
// unusable direct IP. Keep the endpoint reachable, but never pass the
|
||||
// rejected candidate through to iroh's parallel dialer.
|
||||
let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap();
|
||||
let unsafe_socket = "169.254.169.254:80".parse().unwrap();
|
||||
let token = endpoint_token_for_test([
|
||||
TransportAddr::Relay(good_relay.clone()),
|
||||
TransportAddr::Ip(unsafe_socket),
|
||||
]);
|
||||
let validated =
|
||||
validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap();
|
||||
let payload = URL_SAFE_NO_PAD.decode(validated.join_token).unwrap();
|
||||
let sanitized: EndpointAddr = serde_json::from_slice(&payload).unwrap();
|
||||
assert_eq!(
|
||||
sanitized.addrs,
|
||||
[TransportAddr::Relay(good_relay)].into_iter().collect()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_with_all_junk_candidates_is_rejected() {
|
||||
// Guard: if EVERY candidate is unusable, the endpoint must still fail.
|
||||
let token = endpoint_token_for_test([
|
||||
TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), // port 0
|
||||
TransportAddr::Ip("127.0.0.1:9337".parse().unwrap()), // loopback
|
||||
]);
|
||||
assert!(
|
||||
validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err(),
|
||||
"endpoint with no usable candidate must be rejected"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn default_mode_still_rejects_unknown_relay() {
|
||||
// Guard the fix doesn't over-open: a relay that is neither iroh-prod nor
|
||||
// a mesh-llm default must still be rejected under Default mode.
|
||||
let unknown: RelayUrl = "https://not-a-real-relay.example".parse().unwrap();
|
||||
assert!(!relay_allowed(&unknown, &IrohRelayMode::Default));
|
||||
let token = endpoint_token_for_test([TransportAddr::Relay(unknown)]);
|
||||
assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_endpoint_rejects_unsafe_direct_targets_and_oversized_tokens() {
|
||||
for socket in ["127.0.0.1:9337", "169.254.169.254:80", "0.0.0.0:1"] {
|
||||
@@ -263,10 +421,59 @@ mod tests {
|
||||
.expect("sign test bootstrap token");
|
||||
let token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&signed).unwrap());
|
||||
assert_eq!(
|
||||
validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap(),
|
||||
validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default)
|
||||
.unwrap()
|
||||
.endpoint_id,
|
||||
endpoint.id.to_string()
|
||||
);
|
||||
|
||||
let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap();
|
||||
let placeholder_endpoint = EndpointAddr {
|
||||
id: endpoint.id,
|
||||
addrs: [
|
||||
TransportAddr::Relay(good_relay.clone()),
|
||||
TransportAddr::Ip("180.181.228.108:0".parse().unwrap()),
|
||||
]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
};
|
||||
let placeholder_signed = SignedBootstrapToken::sign(
|
||||
vec![serde_json::to_vec(&placeholder_endpoint).unwrap()],
|
||||
&signed_policy,
|
||||
None,
|
||||
&owner,
|
||||
)
|
||||
.expect("sign placeholder test bootstrap token");
|
||||
let placeholder_token =
|
||||
URL_SAFE_NO_PAD.encode(serde_json::to_vec(&placeholder_signed).unwrap());
|
||||
assert!(
|
||||
validate_advertised_endpoint_with_mode(&placeholder_token, &IrohRelayMode::Default)
|
||||
.is_ok(),
|
||||
"signed stock token with a port-0 placeholder must remain usable"
|
||||
);
|
||||
|
||||
let mixed_endpoint = EndpointAddr {
|
||||
id: endpoint.id,
|
||||
addrs: [
|
||||
TransportAddr::Relay(good_relay),
|
||||
TransportAddr::Ip("169.254.169.254:80".parse().unwrap()),
|
||||
]
|
||||
.into_iter()
|
||||
.collect(),
|
||||
};
|
||||
let mixed_signed = SignedBootstrapToken::sign(
|
||||
vec![serde_json::to_vec(&mixed_endpoint).unwrap()],
|
||||
&signed_policy,
|
||||
None,
|
||||
&owner,
|
||||
)
|
||||
.expect("sign mixed test bootstrap token");
|
||||
let mixed_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&mixed_signed).unwrap());
|
||||
assert!(
|
||||
validate_advertised_endpoint_with_mode(&mixed_token, &IrohRelayMode::Default).is_err(),
|
||||
"signed mixed-candidate tokens must fail closed because they cannot be rewritten"
|
||||
);
|
||||
|
||||
let mut tampered = signed;
|
||||
tampered.serialized_addrs[0] = serde_json::to_vec(&EndpointAddr {
|
||||
id: endpoint.id,
|
||||
|
||||
Reference in New Issue
Block a user