[feat] Basic IPNS support

Author: ppodolsky

iroh-p2p/src/dht_records.rs

@@ -0,0 +1,36 @@
+use ahash::AHashMap;
+use libp2p::kad::{GetRecordOk, GetRecordResult, PeerRecord, QueryId};
+use tokio::sync::mpsc::Sender;
+
+pub struct DhtGetQuery {
+    response_channel: Sender<anyhow::Result<PeerRecord>>,
+}
+
+impl DhtGetQuery {
+    pub fn new(response_channel: Sender<anyhow::Result<PeerRecord>>) -> DhtGetQuery {
+        DhtGetQuery { response_channel }
+    }
+}
+
+#[derive(Default)]
+pub struct DhtRecords {
+    current_queries: AHashMap<QueryId, DhtGetQuery>,
+}
+
+impl DhtRecords {
+    pub fn insert(&mut self, query_id: QueryId, query: DhtGetQuery) {
+        self.current_queries.insert(query_id, query);
+    }
+
+    pub fn handle_get_record_result(&mut self, id: QueryId, get_record_result: GetRecordResult) {
+        if let Some(query) = self.current_queries.remove(&id) {
+            match get_record_result {
+                Ok(GetRecordOk::FoundRecord(record)) => {
+                    tokio::spawn(async move { query.response_channel.send(Ok(record)).await.ok() });
+                }
+                Ok(GetRecordOk::FinishedWithNoAdditionalRecord { cache_candidates }) => {}
+                Err(_) => todo!(),
+            }
+        };
+    }
+}

iroh-p2p/src/lib.rs

@@ -1,6 +1,7 @@
 mod behaviour;
 pub mod cli;
 pub mod config;
+mod dht_records;
 mod keys;
 pub mod metrics;
 mod node;

iroh-p2p/src/node.rs

@@ -34,6 +34,7 @@ use tracing::{debug, error, info, trace, warn};
 use iroh_bitswap::{BitswapEvent, Block};
 use iroh_rpc_client::Lookup;
 
+use crate::dht_records::{DhtGetQuery, DhtRecords};
 use crate::keys::{Keychain, Storage};
 use crate::providers::Providers;
 use crate::rpc::{P2p, ProviderRequestKey};
@@ -87,6 +88,7 @@ pub struct Node<KeyStorage: Storage> {
     use_dht: bool,
     bitswap_sessions: BitswapSessions,
     providers: Providers,
+    dht_records: DhtRecords,
     listen_addrs: Vec<Multiaddr>,
 }
 
@@ -175,6 +177,7 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             use_dht: libp2p_config.kademlia,
             bitswap_sessions: Default::default(),
             providers: Providers::new(4),
+            dht_records: DhtRecords::default(),
             listen_addrs,
         })
     }
@@ -522,7 +525,6 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             }
             Event::Kademlia(e) => {
                 libp2p_metrics().record(&e);
-
                 if let KademliaEvent::OutboundQueryProgressed {
                     id, result, step, ..
                 } = e
@@ -627,6 +629,11 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
                                 });
                             }
                         }
+                        QueryResult::GetRecord(get_record_result) => self
+                            .dht_records
+                            .handle_get_record_result(id, get_record_result),
+                        QueryResult::PutRecord(_) => todo!(),
+                        QueryResult::RepublishRecord(_) => todo!(),
                         other => {
                             debug!("Libp2p => Unhandled Kademlia query result: {:?}", other)
                         }
@@ -1061,6 +1068,16 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             RpcMessage::Shutdown => {
                 return Ok(true);
             }
+            RpcMessage::DhtGet {
+                key,
+                response_channel,
+            } => {
+                if let Some(kad) = self.swarm.behaviour_mut().kad.as_mut() {
+                    let query_id = kad.get_record(key);
+                    self.dht_records
+                        .insert(query_id, DhtGetQuery::new(response_channel));
+                }
+            }
         }
 
         Ok(false)

iroh-p2p/src/rpc.rs

@@ -27,6 +27,20 @@ use tokio::sync::mpsc::{channel, Sender};
 use tokio::sync::oneshot;
 use tracing::{debug, info, trace};
 
+use async_trait::async_trait;
+use iroh_bitswap::Block;
+use iroh_rpc_client::Lookup;
+use iroh_rpc_types::p2p::{
+    BitswapRequest, BitswapResponse, ConnectByPeerIdRequest, ConnectRequest, DhtGetRequest,
+    DisconnectRequest, GetListeningAddrsResponse, GetPeersResponse, GossipsubAllPeersResponse,
+    GossipsubPeerAndTopics, GossipsubPeerIdMsg, GossipsubPeersResponse, GossipsubPublishRequest,
+    GossipsubPublishResponse, GossipsubSubscribeResponse, GossipsubTopicHashMsg,
+    GossipsubTopicsResponse, Key as ProviderKey, LookupRequest, Multiaddrs,
+    NotifyNewBlocksBitswapRequest, P2p as RpcP2p, P2pServerAddr, PeerIdResponse, PeerInfo,
+    Providers, StopSessionBitswapRequest, VersionResponse,
+};
+use libp2p::kad::PeerRecord;
+
 use super::node::DEFAULT_PROVIDER_LIMIT;
 use crate::VERSION;
 
@@ -231,6 +245,24 @@ impl P2p {
         Ok(stream)
     }
 
+    #[tracing::instrument(skip(self, req))]
+    async fn dht_get(
+        &self,
+        req: DhtGetRequest,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<iroh_rpc_types::p2p::PeerRecord>> + Send>>> {
+        // ToDo: parametrize size of channel in a proper way
+        let (sender, receiver) = channel(64);
+        let msg = RpcMessage::DhtGet {
+            key: Key::from(req.key.unwrap().key),
+            response_channel: sender,
+        };
+        self.sender.send(msg).await?;
+        let receiver = tokio_stream::wrappers::ReceiverStream::new(receiver)
+            .map(|peer_record| peer_record.map(|peer_record| peer_record.into()));
+
+        Ok(Box::pin(receiver))
+    }
+
     #[tracing::instrument(skip(self, req))]
     async fn start_providing(self, req: StartProvidingRequest) -> Result<()> {
         trace!("received StartProviding request: {:?}", req.key);
@@ -620,6 +652,10 @@ pub enum RpcMessage {
     ExternalAddrs(oneshot::Sender<Vec<Multiaddr>>),
     Listeners(oneshot::Sender<Vec<Multiaddr>>),
     LocalPeerId(oneshot::Sender<PeerId>),
+    DhtGet {
+        key: Key,
+        response_channel: Sender<Result<PeerRecord>>,
+    },
     BitswapRequest {
         ctx: u64,
         cids: Vec<Cid>,

iroh-resolver/src/ipns.proto

@@ -0,0 +1,34 @@
+syntax = "proto2";
+
+package ipns_pb;
+
+message IpnsEntry {
+	enum ValidityType {
+		// setting an EOL says "this record is valid until..."
+		EOL = 0;
+	}
+	optional bytes value = 1;
+	optional bytes signatureV1 = 2;
+
+	optional ValidityType validityType = 3;
+	optional bytes validity = 4;
+
+	optional uint64 sequence = 5;
+
+	optional uint64 ttl = 6;
+
+	// in order for nodes to properly validate a record upon receipt, they need the public
+	// key associated with it. For old RSA keys, its easiest if we just send this as part of
+	// the record itself. For newer ed25519 keys, the public key can be embedded in the
+	// peerID, making this field unnecessary.
+	optional bytes pubKey = 7;
+
+	optional bytes signatureV2 = 8;
+
+	optional bytes data = 9;
+}
+
+message IpnsSignatureV2Checker {
+	optional bytes pubKey = 7;
+	optional bytes signatureV2 = 8;
+}