Compare commits

...

4 Commits

Author SHA1 Message Date
link2xt
545df8c7a3 chore: update Cargo.lock 2025-10-26 17:35:53 +00:00
link2xt
ea6437e98e build: update iroh to 0.94
https://www.iroh.computer/blog/iroh-0-94-0-the-endpoint-takeover
2025-10-26 17:35:53 +00:00
link2xt
c02a8fb219 chore: update Cargo.lock 2025-10-26 17:35:53 +00:00
link2xt
177de13659 build: update iroh to 0.92.0 2025-10-26 17:35:53 +00:00
8 changed files with 1071 additions and 949 deletions

1805
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -66,8 +66,8 @@ humansize = "2"
hyper = "1"
hyper-util = "0.1.16"
image = { version = "0.25.6", default-features=false, features = ["gif", "jpeg", "ico", "png", "pnm", "webp", "bmp"] }
iroh-gossip = { version = "0.35", default-features = false, features = ["net"] }
iroh = { version = "0.35", default-features = false }
iroh-gossip = { version = "0.94", default-features = false, features = ["net"] }
iroh = { version = "0.94", default-features = false }
kamadak-exif = "0.6.1"
libc = { workspace = true }
mail-builder = { version = "0.4.4", default-features = false }

View File

@@ -7,6 +7,9 @@ ignore = [
# <https://rustsec.org/advisories/RUSTSEC-2023-0071>
"RUSTSEC-2023-0071",
# Archived repository
"RUSTSEC-2023-0089",
# Unmaintained instant
"RUSTSEC-2024-0384",
@@ -26,21 +29,20 @@ skip = [
{ name = "derive_more", version = "1.0.0" },
{ name = "event-listener", version = "2.5.3" },
{ name = "getrandom", version = "0.2.12" },
{ name = "hashbrown", version = "0.14.5" },
{ name = "heck", version = "0.4.1" },
{ name = "http", version = "0.2.12" },
{ name = "linux-raw-sys", version = "0.4.14" },
{ name = "lru", version = "0.12.3" },
{ name = "netlink-packet-route", version = "0.17.1" },
{ name = "netdev", version = "0.36.0" },
{ name = "netlink-packet-route", version = "0.22.0" },
{ name = "nom", version = "7.1.3" },
{ name = "rand_chacha", version = "0.3.1" },
{ name = "rand_core", version = "0.6.4" },
{ name = "rand", version = "0.8.5" },
{ name = "rustix", version = "0.38.44" },
{ name = "rustls-webpki", version = "0.102.8" },
{ name = "serdect", version = "0.2.0" },
{ name = "socket2", version = "0.5.9" },
{ name = "spin", version = "0.9.8" },
{ name = "strum_macros", version = "0.26.2" },
{ name = "strum", version = "0.26.2" },
{ name = "syn", version = "1.0.109" },
{ name = "thiserror-impl", version = "1.0.69" },
{ name = "thiserror", version = "1.0.69" },
@@ -55,6 +57,7 @@ skip = [
{ name = "windows_i686_msvc" },
{ name = "windows-implement" },
{ name = "windows-interface" },
{ name = "windows-link" },
{ name = "windows-result" },
{ name = "windows-strings" },
{ name = "windows-sys" },

View File

@@ -69,7 +69,7 @@ pub struct BackupProvider {
_endpoint: Endpoint,
/// iroh address.
node_addr: iroh::NodeAddr,
node_addr: iroh::EndpointAddr,
/// Authentication token that should be submitted
/// to retrieve the backup.
@@ -96,12 +96,11 @@ impl BackupProvider {
pub async fn prepare(context: &Context) -> Result<Self> {
let relay_mode = RelayMode::Disabled;
let endpoint = Endpoint::builder()
.tls_x509() // For compatibility with iroh <0.34.0
.alpns(vec![BACKUP_ALPN.to_vec()])
.relay_mode(relay_mode)
.bind()
.await?;
let node_addr = endpoint.node_addr().await?;
let node_addr = endpoint.addr();
// Acquire global "ongoing" mutex.
let cancel_token = context.alloc_ongoing().await?;
@@ -298,16 +297,12 @@ impl Future for BackupProvider {
pub async fn get_backup2(
context: &Context,
node_addr: iroh::NodeAddr,
node_addr: iroh::EndpointAddr,
auth_token: String,
) -> Result<()> {
let relay_mode = RelayMode::Disabled;
let endpoint = Endpoint::builder()
.tls_x509() // For compatibility with iroh <0.34.0
.relay_mode(relay_mode)
.bind()
.await?;
let endpoint = Endpoint::builder().relay_mode(relay_mode).bind().await?;
let conn = endpoint.connect(node_addr, BACKUP_ALPN).await?;
let (mut send_stream, mut recv_stream) = conn.open_bi().await?;

View File

@@ -1545,7 +1545,7 @@ impl MimeFactory {
// We should not send `null` as relay URL
// as this is the only way to reach the node.
debug_assert!(node_addr.relay_url().is_some());
debug_assert_eq!(node_addr.relay_urls().count(), 1);
headers.push((
HeaderDef::IrohNodeAddr.into(),
mail_builder::headers::text::Text::new(serde_json::to_string(&node_addr)?)

View File

@@ -19,18 +19,22 @@
//! This message contains the users relay-server and public key.
//! Direct IP address is not included as this information can be persisted by email providers.
//! 4. After the announcement, the sending peer joins the gossip swarm with an empty list of peer IDs (as they don't know anyone yet).
//! 5. Upon receiving an announcement message, other peers store the sender's [NodeAddr] in the database
//! 5. Upon receiving an announcement message, other peers store the sender's [EndpointAddr] in the database
//! (scoped per WebXDC app instance/message-id). The other peers can then join the gossip with `joinRealtimeChannel().setListener()`
//! and `joinRealtimeChannel().send()` just like the other peers.
use anyhow::{Context as _, Result, anyhow, bail};
use data_encoding::BASE32_NOPAD;
use futures_lite::StreamExt;
use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMode, RelayUrl, SecretKey};
use iroh_gossip::net::{Event, GOSSIP_ALPN, Gossip, GossipEvent, JoinOptions};
use iroh::discovery::static_provider::StaticProvider;
use iroh::{
Endpoint, EndpointAddr, EndpointId, PublicKey, RelayMode, RelayUrl, SecretKey, TransportAddr,
};
use iroh_gossip::api::{Event as GossipEvent, GossipReceiver, GossipSender, JoinOptions};
use iroh_gossip::net::{GOSSIP_ALPN, Gossip};
use iroh_gossip::proto::TopicId;
use parking_lot::Mutex;
use std::collections::{BTreeSet, HashMap};
use std::collections::HashMap;
use std::env;
use tokio::sync::{RwLock, oneshot};
use tokio::task::JoinHandle;
@@ -54,6 +58,9 @@ pub struct Iroh {
/// Iroh router needed for Iroh peer channels.
pub(crate) router: iroh::protocol::Router,
/// Discovery service.
pub(crate) discovery: StaticProvider,
/// [Gossip] needed for Iroh peer channels.
pub(crate) gossip: Gossip,
@@ -105,7 +112,7 @@ impl Iroh {
}
let peers = get_iroh_gossip_peers(ctx, msg_id).await?;
let node_ids = peers.iter().map(|p| p.node_id).collect::<Vec<_>>();
let node_ids = peers.iter().map(|p| p.id).collect::<Vec<_>>();
info!(
ctx,
@@ -115,7 +122,7 @@ impl Iroh {
// Inform iroh of potentially new node addresses
for node_addr in &peers {
if !node_addr.is_empty() {
self.router.endpoint().add_node_addr(node_addr.clone())?;
self.discovery.add_endpoint_info(node_addr.clone());
}
}
@@ -124,6 +131,7 @@ impl Iroh {
let (gossip_sender, gossip_receiver) = self
.gossip
.subscribe_with_opts(topic, JoinOptions::with_bootstrap(node_ids))
.await?
.split();
let ctx = ctx.clone();
@@ -139,10 +147,10 @@ impl Iroh {
}
/// Add gossip peer to realtime channel if it is already active.
pub async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
pub async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: EndpointAddr) -> Result<()> {
if self.iroh_channels.read().await.get(&topic).is_some() {
self.router.endpoint().add_node_addr(peer.clone())?;
self.gossip.subscribe(topic, vec![peer.node_id])?;
self.discovery.add_endpoint_info(peer.clone());
self.gossip.subscribe(topic, vec![peer.id]).await?;
}
Ok(())
}
@@ -184,16 +192,20 @@ impl Iroh {
*entry
}
/// Get the iroh [NodeAddr] without direct IP addresses.
/// Get the iroh [EndpointAddr] without direct IP addresses.
///
/// The address is guaranteed to have home relay URL set
/// as it is the only way to reach the node
/// without global discovery mechanisms.
pub(crate) async fn get_node_addr(&self) -> Result<NodeAddr> {
let mut addr = self.router.endpoint().node_addr().await?;
addr.direct_addresses = BTreeSet::new();
debug_assert!(addr.relay_url().is_some());
Ok(addr)
pub(crate) async fn get_node_addr(&self) -> Result<EndpointAddr> {
// Wait until home relay connection is established.
self.router.endpoint().online().await;
let mut endpoint_addr = self.router.endpoint().addr();
endpoint_addr
.addrs
.retain(|addr| matches!(addr, TransportAddr::Relay(_)));
debug_assert_eq!(endpoint_addr.addrs.len(), 1);
Ok(endpoint_addr)
}
/// Leave the realtime channel for a given topic.
@@ -219,11 +231,11 @@ pub(crate) struct ChannelState {
/// The subscribe loop handle.
subscribe_loop: JoinHandle<()>,
sender: iroh_gossip::net::GossipSender,
sender: GossipSender,
}
impl ChannelState {
fn new(subscribe_loop: JoinHandle<()>, sender: iroh_gossip::net::GossipSender) -> Self {
fn new(subscribe_loop: JoinHandle<()>, sender: GossipSender) -> Self {
Self {
subscribe_loop,
sender,
@@ -235,7 +247,7 @@ impl Context {
/// Create iroh endpoint and gossip.
async fn init_peer_channels(&self) -> Result<Iroh> {
info!(self, "Initializing peer channels.");
let secret_key = SecretKey::generate(rand_old::rngs::OsRng);
let secret_key = SecretKey::generate(&mut rand::rng());
let public_key = secret_key.public();
let relay_mode = if let Some(relay_url) = self
@@ -252,8 +264,9 @@ impl Context {
RelayMode::Default
};
let discovery = StaticProvider::new();
let endpoint = Endpoint::builder()
.tls_x509() // For compatibility with iroh <0.34.0
.discovery(discovery.clone())
.secret_key(secret_key)
.alpns(vec![GOSSIP_ALPN.to_vec()])
.relay_mode(relay_mode)
@@ -267,8 +280,7 @@ impl Context {
let gossip = Gossip::builder()
.max_message_size(128 * 1024)
.spawn(endpoint.clone())
.await?;
.spawn(endpoint.clone());
let router = iroh::protocol::Router::builder(endpoint)
.accept(GOSSIP_ALPN, gossip.clone())
@@ -276,6 +288,7 @@ impl Context {
Ok(Iroh {
router,
discovery,
gossip,
sequence_numbers: Mutex::new(HashMap::new()),
iroh_channels: RwLock::new(HashMap::new()),
@@ -322,11 +335,15 @@ impl Context {
}
}
pub(crate) async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
pub(crate) async fn maybe_add_gossip_peer(
&self,
topic: TopicId,
peer: EndpointAddr,
) -> Result<()> {
if let Some(iroh) = &*self.iroh.read().await {
info!(
self,
"Adding (maybe existing) peer with id {} to {topic}.", peer.node_id
"Adding (maybe existing) peer with id {} to {topic}.", peer.id
);
iroh.maybe_add_gossip_peer(topic, peer).await?;
}
@@ -334,12 +351,12 @@ impl Context {
}
}
/// Cache a peers [NodeId] for one topic.
/// Cache a peers [EndpointId] for one topic.
pub(crate) async fn iroh_add_peer_for_topic(
ctx: &Context,
msg_id: MsgId,
topic: TopicId,
peer: NodeId,
peer: EndpointId,
relay_server: Option<&str>,
) -> Result<()> {
ctx.sql
@@ -365,11 +382,11 @@ pub async fn add_gossip_peer_from_header(
}
let node_addr =
serde_json::from_str::<NodeAddr>(node_addr).context("Failed to parse node address")?;
serde_json::from_str::<EndpointAddr>(node_addr).context("Failed to parse node address")?;
info!(
context,
"Adding iroh peer with node id {} to the topic of {instance_id}.", node_addr.node_id
"Adding iroh peer with node id {} to the topic of {instance_id}.", node_addr.id
);
context.emit_event(EventType::WebxdcRealtimeAdvertisementReceived {
@@ -384,8 +401,8 @@ pub async fn add_gossip_peer_from_header(
return Ok(());
};
let node_id = node_addr.node_id;
let relay_server = node_addr.relay_url().map(|relay| relay.as_str());
let node_id = node_addr.id;
let relay_server = node_addr.relay_urls().map(|relay| relay.as_str()).next();
iroh_add_peer_for_topic(context, instance_id, topic, node_id, relay_server).await?;
context.maybe_add_gossip_peer(topic, node_addr).await?;
@@ -403,8 +420,8 @@ pub(crate) async fn insert_topic_stub(ctx: &Context, msg_id: MsgId, topic: Topic
Ok(())
}
/// Get a list of [NodeAddr]s for one webxdc.
async fn get_iroh_gossip_peers(ctx: &Context, msg_id: MsgId) -> Result<Vec<NodeAddr>> {
/// Get a list of [EndpointAddr]s for one webxdc.
async fn get_iroh_gossip_peers(ctx: &Context, msg_id: MsgId) -> Result<Vec<EndpointAddr>> {
ctx.sql
.query_map(
"SELECT public_key, relay_server FROM iroh_gossip_peers WHERE msg_id = ? AND public_key != ?",
@@ -417,11 +434,11 @@ async fn get_iroh_gossip_peers(ctx: &Context, msg_id: MsgId) -> Result<Vec<NodeA
|g| {
g.map(|data| {
let (key, server) = data?;
let server = server.map(|data| Ok::<_, url::ParseError>(RelayUrl::from(Url::parse(&data)?))).transpose()?;
let id = NodeId::from_bytes(&key.try_into()
let server: Option<TransportAddr> = server.map(|data| Ok::<_, url::ParseError>(TransportAddr::Relay(RelayUrl::from(Url::parse(&data)?)))).transpose()?;
let id = EndpointId::from_bytes(&key.try_into()
.map_err(|_| anyhow!("Can't convert sql data to [u8; 32]"))?)?;
Ok::<_, anyhow::Error>(NodeAddr::from_parts(
id, server, vec![]
Ok::<_, anyhow::Error>(EndpointAddr::from_parts(
id, server
))
})
.collect::<std::result::Result<Vec<_>, _>>()
@@ -523,45 +540,39 @@ pub(crate) async fn create_iroh_header(ctx: &Context, msg_id: MsgId) -> Result<S
async fn subscribe_loop(
context: &Context,
mut stream: iroh_gossip::net::GossipReceiver,
mut stream: GossipReceiver,
topic: TopicId,
msg_id: MsgId,
join_tx: oneshot::Sender<()>,
) -> Result<()> {
let mut join_tx = Some(join_tx);
stream.joined().await?;
// Try to notify that at least one peer joined,
// but ignore the error if receiver is dropped and nobody listens.
join_tx.send(()).ok();
for node in stream.neighbors() {
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
}
while let Some(event) = stream.try_next().await? {
match event {
Event::Gossip(event) => match event {
GossipEvent::Joined(nodes) => {
if let Some(join_tx) = join_tx.take() {
// Try to notify that at least one peer joined,
// but ignore the error if receiver is dropped and nobody listens.
join_tx.send(()).ok();
}
for node in nodes {
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
}
}
GossipEvent::NeighborUp(node) => {
info!(context, "IROH_REALTIME: NeighborUp: {}", node.to_string());
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
}
GossipEvent::NeighborDown(_node) => {}
GossipEvent::Received(message) => {
info!(context, "IROH_REALTIME: Received realtime data");
context.emit_event(EventType::WebxdcRealtimeData {
msg_id,
data: message
.content
.get(0..message.content.len() - 4 - PUBLIC_KEY_LENGTH)
.context("too few bytes in iroh message")?
.into(),
});
}
},
Event::Lagged => {
GossipEvent::NeighborUp(node) => {
info!(context, "IROH_REALTIME: NeighborUp: {}", node.to_string());
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
}
GossipEvent::NeighborDown(_node) => {}
GossipEvent::Received(message) => {
info!(context, "IROH_REALTIME: Received realtime data");
context.emit_event(EventType::WebxdcRealtimeData {
msg_id,
data: message
.content
.get(0..message.content.len() - 4 - PUBLIC_KEY_LENGTH)
.context("too few bytes in iroh message")?
.into(),
});
}
GossipEvent::Lagged => {
warn!(context, "Gossip lost some messages");
}
};
@@ -626,7 +637,7 @@ mod tests {
.await
.unwrap()
.into_iter()
.map(|addr| addr.node_id)
.map(|addr| addr.id)
.collect::<Vec<_>>();
assert_eq!(
@@ -639,7 +650,7 @@ mod tests {
.get_node_addr()
.await
.unwrap()
.node_id
.id
]
);
@@ -702,7 +713,7 @@ mod tests {
.await
.unwrap()
.into_iter()
.map(|addr| addr.node_id)
.map(|addr| addr.id)
.collect::<Vec<_>>();
assert_eq!(
@@ -714,7 +725,7 @@ mod tests {
.get_node_addr()
.await
.unwrap()
.node_id
.id
]
);
@@ -792,7 +803,7 @@ mod tests {
.await
.unwrap()
.into_iter()
.map(|addr| addr.node_id)
.map(|addr| addr.id)
.collect::<Vec<_>>();
assert_eq!(
@@ -805,7 +816,7 @@ mod tests {
.get_node_addr()
.await
.unwrap()
.node_id
.id
]
);

View File

@@ -112,7 +112,7 @@ pub enum Qr {
/// Provides a backup that can be retrieved using iroh-net based backup transfer protocol.
Backup2 {
/// Iroh node address.
node_addr: iroh::NodeAddr,
node_addr: iroh::EndpointAddr,
/// Authentication token.
auth_token: String,
@@ -629,7 +629,7 @@ fn decode_backup2(qr: &str) -> Result<Qr> {
.split_once('&')
.context("Backup QR code has no separator")?;
let auth_token = auth_token.to_string();
let node_addr = serde_json::from_str::<iroh::NodeAddr>(node_addr)
let node_addr = serde_json::from_str::<iroh::EndpointAddr>(node_addr)
.context("Invalid node addr in backup QR code")?;
Ok(Qr::Backup2 {

View File

@@ -865,25 +865,3 @@ async fn test_decode_socks5() -> Result<()> {
Ok(())
}
/// Ensure that `DCBACKUP2` QR code does not fail to deserialize
/// because iroh changes the format of `NodeAddr`
/// as happened between iroh 0.29 and iroh 0.30 before.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_decode_backup() -> Result<()> {
let ctx = TestContext::new().await;
let qr = check_qr(&ctx, r#"DCBACKUP2:TWSv6ZjDPa5eoxkocj7xMi8r&{"node_id":"9afc1ea5b4f543e5cdd7b7a21cd26aee7c0b1e1c2af26790896fbd8932a06e1e","relay_url":null,"direct_addresses":["192.168.1.10:12345"]}"#).await?;
assert!(matches!(qr, Qr::Backup2 { .. }));
let qr = check_qr(&ctx, r#"DCBACKUP2:AIvFjRFBt_aMiisSZ8P33JqY&{"node_id":"buzkyd4x76w66qtanjk5fm6ikeuo4quletajowsl3a3p7l6j23pa","info":{"relay_url":null,"direct_addresses":["192.168.1.5:12345"]}}"#).await?;
assert!(matches!(qr, Qr::Backup2 { .. }));
let qr = check_qr(&ctx, r#"DCBACKUP9:from-the-future"#).await?;
assert!(matches!(qr, Qr::BackupTooNew { .. }));
let qr = check_qr(&ctx, r#"DCBACKUP99:far-from-the-future"#).await?;
assert!(matches!(qr, Qr::BackupTooNew { .. }));
Ok(())
}