Cargo.toml

@@ -2,7 +2,7 @@
 name = "discv5"
 authors = ["Sigma Prime <contact@sigmaprime.io>"]
 edition = "2018"
-version = "0.8.0"
+version = "0.9.1"
 description = "Implementation of the p2p discv5 discovery protocol"
 license = "Apache-2.0"
 repository = "https://github.com/sigp/discv5"
@@ -12,17 +12,20 @@ categories = ["network-programming", "asynchronous"]
 exclude = [".gitignore", ".github/*"]
 
 [dependencies]
-enr = { version = "0.12", features = ["k256", "ed25519"] }
+enr = { version = "0.13.0", features = [
+  "k256",
+  "ed25519",
+] } # enr = { version = "0.12", features = ["k256", "ed25519"] }
 tokio = { version = "1", features = ["net", "sync", "macros", "rt"] }
 libp2p-identity = { version = "0.2", features = [
-    "ed25519",
-    "secp256k1",
+  "ed25519",
+  "secp256k1",
 ], optional = true }
 multiaddr = { version = "0.18", optional = true }
 zeroize = { version = "1", features = ["zeroize_derive"] }
 futures = "0.3"
 uint = { version = "0.10", default-features = false }
-alloy-rlp = { version = "0.3.8", default-features = true }
+alloy-rlp = { version = "0.3", default-features = true }
 # This version must be kept up to date do it uses the same dependencies as ENR
 hkdf = "0.12"
 hex = "0.4"
@@ -33,8 +36,8 @@ socket2 = "0.5"
 smallvec = "1"
 parking_lot = "0.12"
 lazy_static = "1"
-aes = "0.8.4"
-ctr = "0.9.2"
+aes = "0.8"
+ctr = "0.9"
 aes-gcm = "0.10.3"
 tracing = { version = "0.1", features = ["log"] }
 lru = "0.12"

src/config.rs

@@ -15,7 +15,7 @@ pub struct Config {
     pub request_timeout: Duration,
 
     /// The interval over which votes are remembered when determining our external IP. A lower
-    /// interval will respond faster to IP changes. Default is 30 seconds.
+    /// interval will respond faster to IP changes. Default is 2 minutes.
     pub vote_duration: Duration,
 
     /// The timeout after which a `QueryPeer` in an ongoing query is marked unresponsive.
@@ -92,6 +92,18 @@ pub struct Config {
     /// will last indefinitely. Default is 1 hour.
     pub ban_duration: Option<Duration>,
 
+    /// Auto-discovering our IP address, is only one part in discovering our NAT/firewall
+    /// situation. We need to determine if we are behind a firewall that is preventing incoming
+    /// connections (this is especially true for IPv6 where all connections will report the same
+    /// external IP). To do this, Discv5 uses a heuristic, which is that after we set an address in
+    /// our ENR, we wait for this duration to see if we have any incoming connections. If we
+    /// receive a single INCOMING connection in this duration, we consider ourselves contactable,
+    /// until we update or change our IP address again. If we fail to receive an incoming
+    /// connection in this duration, we revoke our ENR address advertisement for 6 hours, before
+    /// trying again. This can be set to None, to always advertise and never revoke. The default is
+    /// Some(5 minutes).
+    pub auto_nat_listen_duration: Option<Duration>,
+
     /// A custom executor which can spawn the discv5 tasks. This must be a tokio runtime, with
     /// timing support. By default, the executor that created the discv5 struct will be used.
     pub executor: Option<Box<dyn Executor + Send + Sync>>,
@@ -121,7 +133,7 @@ impl ConfigBuilder {
         let config = Config {
             enable_packet_filter: false,
             request_timeout: Duration::from_secs(1),
-            vote_duration: Duration::from_secs(30),
+            vote_duration: Duration::from_secs(120),
             query_peer_timeout: Duration::from_secs(2),
             query_timeout: Duration::from_secs(60),
             request_retries: 1,
@@ -141,6 +153,7 @@ impl ConfigBuilder {
             filter_max_bans_per_ip: Some(5),
             permit_ban_list: PermitBanList::default(),
             ban_duration: Some(Duration::from_secs(3600)), // 1 hour
+            auto_nat_listen_duration: Some(Duration::from_secs(300)), // 5 minutes
             executor: None,
             listen_config,
         };
@@ -161,7 +174,7 @@ impl ConfigBuilder {
     }
 
     /// The interval over which votes are remembered when determining our external IP. A lower
-    /// interval will respond faster to IP changes. Default is 30 seconds.
+    /// interval will respond faster to IP changes. Default is 2 minutes.
     pub fn vote_duration(&mut self, vote_duration: Duration) -> &mut Self {
         self.config.vote_duration = vote_duration;
         self
@@ -295,6 +308,24 @@ impl ConfigBuilder {
         self
     }
 
+    /// Auto-discovering our IP address, is only one part in discovering our NAT/firewall
+    /// situation. We need to determine if we are behind a firewall that is preventing incoming
+    /// connections (this is especially true for IPv6 where all connections will report the same
+    /// external IP). To do this, Discv5 uses a heuristic, which is that after we set an address in
+    /// our ENR, we wait for this duration to see if we have any incoming connections. If we
+    /// receive a single INCOMING connection in this duration, we consider ourselves contactable,
+    /// until we update or change our IP address again. If we fail to receive an incoming
+    /// connection in this duration, we revoke our ENR address advertisement for 6 hours, before
+    /// trying again. This can be set to None, to always advertise and never revoke. The default is
+    /// Some(5 minutes).
+    pub fn auto_nat_listen_duration(
+        &mut self,
+        auto_nat_listen_duration: Option<Duration>,
+    ) -> &mut Self {
+        self.config.auto_nat_listen_duration = auto_nat_listen_duration;
+        self
+    }
+
     /// A custom executor which can spawn the discv5 tasks. This must be a tokio runtime, with
     /// timing support.
     pub fn executor(&mut self, executor: Box<dyn Executor + Send + Sync>) -> &mut Self {

src/discv5.rs

@@ -267,6 +267,10 @@ impl<P: ProtocolIdentity> Discv5<P> {
         nodes_to_send
     }
 
+    pub fn ip_mode(&self) -> IpMode {
+        self.ip_mode
+    }
+
     /// Mark a node in the routing table as `Disconnected`.
     ///
     /// A `Disconnected` node will be present in the routing table and will be only
@@ -568,21 +572,19 @@ impl<P: ProtocolIdentity> Discv5<P> {
         }
     }
 
-    /// Request a TALK message from a node, identified via the ENR.
+    /// Request a TALK message from a node, identified via the NodeContact.
     pub fn talk_req(
         &self,
-        enr: Enr,
+        node_contact: NodeContact,
         protocol: Vec<u8>,
         request: Vec<u8>,
     ) -> impl Future<Output = Result<Vec<u8>, RequestError>> + 'static {
         // convert the ENR to a node_contact.
 
         let (callback_send, callback_recv) = oneshot::channel();
         let channel = self.clone_channel();
-        let ip_mode = self.ip_mode;
 
         async move {
-            let node_contact = NodeContact::try_from_enr(enr, ip_mode)?;
             let channel = channel.map_err(|_| RequestError::ServiceNotStarted)?;
 
             let event = ServiceRequest::Talk(node_contact, protocol, request, callback_send);

src/handler/mod.rs

@@ -858,7 +858,7 @@ impl Handler {
                         //
                         // We still handle the request, but we do not add the ENR to our routing
                         // table or consider the ENR valid.
-                        warn!(
+                        debug!(
                             udp4_socket = ?enr.udp4_socket(),
                             udp6_socket = ?enr.udp6_socket(),
                             expected = %node_address,

src/kbucket/bucket.rs

@@ -410,9 +410,7 @@ where
             // Adjust `first_connected_pos` accordingly.
             match old_status.state {
                 ConnectionState::Connected => {
-                    if self.first_connected_pos.map_or(false, |p| p == pos.0)
-                        && pos.0 == self.nodes.len()
-                    {
+                    if self.first_connected_pos == Some(pos.0) && pos.0 == self.nodes.len() {
                         // It was the last connected node.
                         self.first_connected_pos = None
                     }

src/metrics.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 
 lazy_static! {
     pub static ref METRICS: InternalMetrics = InternalMetrics::default();
@@ -16,6 +16,10 @@ pub struct InternalMetrics {
     pub bytes_sent: AtomicUsize,
     /// The number of bytes received.
     pub bytes_recv: AtomicUsize,
+    /// Whether we consider ourselves contactable or not on ipv4.
+    pub ipv4_contactable: AtomicBool,
+    /// Whether we consider ourselves contactable or not on ipv6.
+    pub ipv6_contactable: AtomicBool,
 }
 
 impl Default for InternalMetrics {
@@ -26,6 +30,8 @@ impl Default for InternalMetrics {
             unsolicited_requests_per_window: AtomicUsize::new(0),
             bytes_sent: AtomicUsize::new(0),
             bytes_recv: AtomicUsize::new(0),
+            ipv4_contactable: AtomicBool::new(false),
+            ipv6_contactable: AtomicBool::new(false),
         }
     }
 }
@@ -55,6 +61,10 @@ pub struct Metrics {
     pub bytes_sent: usize,
     /// The number of bytes received.
     pub bytes_recv: usize,
+    /// Whether we consider ourselves contactable or not.
+    pub ipv4_contactable: bool,
+    /// Whether we consider ourselves contactable or not.
+    pub ipv6_contactable: bool,
 }
 
 impl From<&METRICS> for Metrics {
@@ -67,6 +77,8 @@ impl From<&METRICS> for Metrics {
                 / internal_metrics.moving_window as f64,
             bytes_sent: internal_metrics.bytes_sent.load(Ordering::Relaxed),
             bytes_recv: internal_metrics.bytes_recv.load(Ordering::Relaxed),
+            ipv4_contactable: internal_metrics.ipv4_contactable.load(Ordering::Relaxed),
+            ipv6_contactable: internal_metrics.ipv6_contactable.load(Ordering::Relaxed),
         }
     }
 }

src/node_info.rs

@@ -26,8 +26,19 @@ pub struct NonContactable {
 }
 
 impl NodeContact {
+    pub fn new(public_key: CombinedPublicKey, socket_addr: SocketAddr, enr: Option<Enr>) -> Self {
+        NodeContact {
+            public_key,
+            socket_addr,
+            enr,
+        }
+    }
+
     pub fn node_id(&self) -> NodeId {
-        self.public_key.clone().into()
+        match self.enr {
+            Some(ref enr) => enr.node_id(),
+            None => self.public_key.clone().into(),
+        }
     }
 
     pub fn seq_no(&self) -> Option<u64> {

src/rpc.rs

@@ -155,8 +155,8 @@ impl Request {
             RequestBody::Talk { protocol, request } => {
                 let mut list = Vec::<u8>::new();
                 id.as_bytes().encode(&mut list);
-                protocol.encode(&mut list);
-                request.encode(&mut list);
+                protocol.as_slice().encode(&mut list);
+                request.as_slice().encode(&mut list);
                 let header = Header {
                     list: true,
                     payload_length: list.len(),
@@ -479,8 +479,8 @@ impl Message {
             }
             5 => {
                 // Talk Request
-                let protocol = Vec::<u8>::decode(payload)?;
-                let request = Vec::<u8>::decode(payload)?;
+                let protocol = Bytes::decode(payload)?.to_vec();
+                let request = Bytes::decode(payload)?.to_vec();
                 if !payload.is_empty() {
                     return Err(DecoderError::Custom("Payload should be empty"));
                 }
@@ -812,4 +812,50 @@ mod tests {
         let data6 = [6, 193, 128, 128];
         Message::decode(&data6).expect_err("should reject extra data");
     }
+
+    #[test]
+    fn test_encode_request_talk_request() {
+        // reference input
+        let id = RequestId(vec![113, 236, 255, 66, 31, 191, 221, 86]);
+        let message = Message::Request(Request {
+            id,
+            body: RequestBody::Talk {
+                protocol: hex::decode("757470").unwrap(),
+                request: hex::decode("0100a028839e1549000003ef001000007619dde7").unwrap(),
+            },
+        });
+
+        // expected hex output
+        let expected_output =
+            hex::decode("05e28871ecff421fbfdd5683757470940100a028839e1549000003ef001000007619dde7")
+                .unwrap();
+        dbg!(hex::encode(message.clone().encode()));
+
+        let encoded_message = message.clone().encode();
+        assert_eq!(encoded_message.clone(), expected_output);
+        assert_eq!(Message::decode(&encoded_message).unwrap(), message);
+    }
+
+    #[test]
+    fn test_encode_request_talk_response() {
+        // reference input
+        let id = RequestId(vec![113, 236, 255, 66, 31, 191, 221, 86]);
+        let message = Message::Response(Response {
+            id,
+            body: ResponseBody::Talk {
+                response: hex::decode("0100a028839e1549000003ef001000007619dde7").unwrap(),
+            },
+        });
+
+        // expected hex output
+        let expected_output =
+            hex::decode("06de8871ecff421fbfdd56940100a028839e1549000003ef001000007619dde7")
+                .unwrap();
+
+        dbg!(hex::encode(message.clone().encode()));
+
+        let encoded_message = message.clone().encode();
+        assert_eq!(encoded_message.clone(), expected_output);
+        assert_eq!(Message::decode(&encoded_message).unwrap(), message);
+    }
 }