libp2p_identify/
identify.rs

1// Copyright 2018 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21use crate::handler::{IdentifyHandler, IdentifyHandlerEvent};
22use crate::protocol::{IdentifyInfo, ReplySubstream};
23use futures::prelude::*;
24use libp2p_core::{
25    ConnectedPoint,
26    Multiaddr,
27    PeerId,
28    PublicKey,
29    connection::ConnectionId,
30    upgrade::{ReadOneError, UpgradeError}
31};
32use libp2p_swarm::{
33    AddressScore,
34    NegotiatedSubstream,
35    NetworkBehaviour,
36    NetworkBehaviourAction,
37    PollParameters,
38    ProtocolsHandler,
39    ProtocolsHandlerUpgrErr
40};
41use std::{
42    collections::{HashMap, VecDeque},
43    io,
44    pin::Pin,
45    task::Context,
46    task::Poll
47};
48
49/// Network behaviour that automatically identifies nodes periodically, returns information
50/// about them, and answers identify queries from other nodes.
51///
52/// All external addresses of the local node supposedly observed by remotes
53/// are reported via [`NetworkBehaviourAction::ReportObservedAddr`] with a
54/// [score](AddressScore) of `1`.
55pub struct Identify {
56    /// Protocol version to send back to remotes.
57    protocol_version: String,
58    /// Agent version to send back to remotes.
59    agent_version: String,
60    /// The public key of the local node. To report on the wire.
61    local_public_key: PublicKey,
62    /// For each peer we're connected to, the observed address to send back to it.
63    observed_addresses: HashMap<PeerId, HashMap<ConnectionId, Multiaddr>>,
64    /// Pending replies to send.
65    pending_replies: VecDeque<Reply>,
66    /// Pending events to be emitted when polled.
67    events: VecDeque<NetworkBehaviourAction<(), IdentifyEvent>>,
68}
69
70/// A pending reply to an inbound identification request.
71enum Reply {
72    /// The reply is queued for sending.
73    Queued {
74        peer: PeerId,
75        io: ReplySubstream<NegotiatedSubstream>,
76        observed: Multiaddr
77    },
78    /// The reply is being sent.
79    Sending {
80        peer: PeerId,
81        io: Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>>,
82    }
83}
84
85impl Identify {
86    /// Creates a new `Identify` network behaviour.
87    pub fn new(protocol_version: String, agent_version: String, local_public_key: PublicKey) -> Self {
88        Identify {
89            protocol_version,
90            agent_version,
91            local_public_key,
92            observed_addresses: HashMap::new(),
93            pending_replies: VecDeque::new(),
94            events: VecDeque::new(),
95        }
96    }
97}
98
99impl NetworkBehaviour for Identify {
100    type ProtocolsHandler = IdentifyHandler;
101    type OutEvent = IdentifyEvent;
102
103    fn new_handler(&mut self) -> Self::ProtocolsHandler {
104        IdentifyHandler::new()
105    }
106
107    fn addresses_of_peer(&mut self, _: &PeerId) -> Vec<Multiaddr> {
108        Vec::new()
109    }
110
111    fn inject_connected(&mut self, _: &PeerId) {
112    }
113
114    fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
115        let addr = match endpoint {
116            ConnectedPoint::Dialer { address } => address.clone(),
117            ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr.clone(),
118        };
119
120        self.observed_addresses.entry(*peer_id).or_default().insert(*conn, addr);
121    }
122
123    fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, _: &ConnectedPoint) {
124        if let Some(addrs) = self.observed_addresses.get_mut(peer_id) {
125            addrs.remove(conn);
126        }
127    }
128
129    fn inject_disconnected(&mut self, peer_id: &PeerId) {
130        self.observed_addresses.remove(peer_id);
131    }
132
133    fn inject_event(
134        &mut self,
135        peer_id: PeerId,
136        connection: ConnectionId,
137        event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
138    ) {
139        match event {
140            IdentifyHandlerEvent::Identified(remote) => {
141                self.events.push_back(
142                    NetworkBehaviourAction::GenerateEvent(
143                        IdentifyEvent::Received {
144                            peer_id,
145                            info: remote.info,
146                            observed_addr: remote.observed_addr.clone(),
147                        }));
148                self.events.push_back(
149                    NetworkBehaviourAction::ReportObservedAddr {
150                        address: remote.observed_addr,
151                        score: AddressScore::Finite(1),
152                    });
153            }
154            IdentifyHandlerEvent::Identify(sender) => {
155                let observed = self.observed_addresses.get(&peer_id)
156                    .and_then(|addrs| addrs.get(&connection))
157                    .expect("`inject_event` is only called with an established connection \
158                             and `inject_connection_established` ensures there is an entry; qed");
159                self.pending_replies.push_back(
160                    Reply::Queued {
161                        peer: peer_id,
162                        io: sender,
163                        observed: observed.clone()
164                    });
165            }
166            IdentifyHandlerEvent::IdentificationError(error) => {
167                self.events.push_back(
168                    NetworkBehaviourAction::GenerateEvent(
169                        IdentifyEvent::Error { peer_id, error }));
170            }
171        }
172    }
173
174    fn poll(
175        &mut self,
176        cx: &mut Context<'_>,
177        params: &mut impl PollParameters,
178    ) -> Poll<
179        NetworkBehaviourAction<
180            <Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
181            Self::OutEvent,
182        >,
183    > {
184        if let Some(event) = self.events.pop_front() {
185            return Poll::Ready(event);
186        }
187
188        if let Some(r) = self.pending_replies.pop_front() {
189            // The protocol names can be bytes, but the identify protocol except UTF-8 strings.
190            // There's not much we can do to solve this conflict except strip non-UTF-8 characters.
191            let protocols: Vec<_> = params
192                .supported_protocols()
193                .map(|p| String::from_utf8_lossy(&p).to_string())
194                .collect();
195
196            let mut listen_addrs: Vec<_> = params.external_addresses().map(|r| r.addr).collect();
197            listen_addrs.extend(params.listened_addresses());
198
199            let mut sending = 0;
200            let to_send = self.pending_replies.len() + 1;
201            let mut reply = Some(r);
202            loop {
203                match reply {
204                    Some(Reply::Queued { peer, io, observed }) => {
205                        let info = IdentifyInfo {
206                            public_key: self.local_public_key.clone(),
207                            protocol_version: self.protocol_version.clone(),
208                            agent_version: self.agent_version.clone(),
209                            listen_addrs: listen_addrs.clone(),
210                            protocols: protocols.clone(),
211                        };
212                        let io = Box::pin(io.send(info, &observed));
213                        reply = Some(Reply::Sending { peer, io });
214                    }
215                    Some(Reply::Sending { peer, mut io }) => {
216                        sending += 1;
217                        match Future::poll(Pin::new(&mut io), cx) {
218                            Poll::Ready(Ok(())) => {
219                                let event = IdentifyEvent::Sent { peer_id: peer };
220                                return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
221                            },
222                            Poll::Pending => {
223                                self.pending_replies.push_back(Reply::Sending { peer, io });
224                                if sending == to_send {
225                                    // All remaining futures are NotReady
226                                    break
227                                } else {
228                                    reply = self.pending_replies.pop_front();
229                                }
230                            }
231                            Poll::Ready(Err(err)) => {
232                                let event = IdentifyEvent::Error {
233                                    peer_id: peer,
234                                    error: ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(err.into()))
235                                };
236                                return Poll::Ready(NetworkBehaviourAction::GenerateEvent(event));
237                            },
238                        }
239                    }
240                    None => unreachable!()
241                }
242            }
243        }
244
245        Poll::Pending
246    }
247}
248
249/// Event emitted  by the `Identify` behaviour.
250#[derive(Debug)]
251pub enum IdentifyEvent {
252    /// Identifying information has been received from a peer.
253    Received {
254        /// The peer that has been identified.
255        peer_id: PeerId,
256        /// The information provided by the peer.
257        info: IdentifyInfo,
258        /// The address observed by the peer for the local node.
259        observed_addr: Multiaddr,
260    },
261    /// Identifying information of the local node has been sent to a peer.
262    Sent {
263        /// The peer that the information has been sent to.
264        peer_id: PeerId,
265    },
266    /// Error while attempting to identify the remote.
267    Error {
268        /// The peer with whom the error originated.
269        peer_id: PeerId,
270        /// The error that occurred.
271        error: ProtocolsHandlerUpgrErr<ReadOneError>,
272    },
273}
274
275#[cfg(test)]
276mod tests {
277    use crate::{Identify, IdentifyEvent};
278    use futures::{prelude::*, pin_mut};
279    use libp2p_core::{
280        identity,
281        PeerId,
282        muxing::StreamMuxerBox,
283        transport,
284        Transport,
285        upgrade
286    };
287    use libp2p_noise as noise;
288    use libp2p_tcp::TcpConfig;
289    use libp2p_swarm::{Swarm, SwarmEvent};
290    use libp2p_mplex::MplexConfig;
291
292    fn transport() -> (identity::PublicKey, transport::Boxed<(PeerId, StreamMuxerBox)>) {
293        let id_keys = identity::Keypair::generate_ed25519();
294        let noise_keys = noise::Keypair::<noise::X25519Spec>::new().into_authentic(&id_keys).unwrap();
295        let pubkey = id_keys.public();
296        let transport = TcpConfig::new()
297            .nodelay(true)
298            .upgrade(upgrade::Version::V1)
299            .authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
300            .multiplex(MplexConfig::new())
301            .boxed();
302        (pubkey, transport)
303    }
304
305    #[test]
306    fn periodic_id_works() {
307        let (mut swarm1, pubkey1) = {
308            let (pubkey, transport) = transport();
309            let protocol = Identify::new("a".to_string(), "b".to_string(), pubkey.clone());
310            let swarm = Swarm::new(transport, protocol, pubkey.clone().into_peer_id());
311            (swarm, pubkey)
312        };
313
314        let (mut swarm2, pubkey2) = {
315            let (pubkey, transport) = transport();
316            let protocol = Identify::new("c".to_string(), "d".to_string(), pubkey.clone());
317            let swarm = Swarm::new(transport, protocol, pubkey.clone().into_peer_id());
318            (swarm, pubkey)
319        };
320
321        Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
322
323        let listen_addr = async_std::task::block_on(async {
324            loop {
325                let swarm1_fut = swarm1.next_event();
326                pin_mut!(swarm1_fut);
327                match swarm1_fut.await {
328                    SwarmEvent::NewListenAddr(addr) => return addr,
329                    _ => {}
330                }
331            }
332        });
333        Swarm::dial_addr(&mut swarm2, listen_addr).unwrap();
334
335        // nb. Either swarm may receive the `Identified` event first, upon which
336        // it will permit the connection to be closed, as defined by
337        // `IdentifyHandler::connection_keep_alive`. Hence the test succeeds if
338        // either `Identified` event arrives correctly.
339        async_std::task::block_on(async move {
340            loop {
341                let swarm1_fut = swarm1.next();
342                pin_mut!(swarm1_fut);
343                let swarm2_fut = swarm2.next();
344                pin_mut!(swarm2_fut);
345
346                match future::select(swarm1_fut, swarm2_fut).await.factor_second().0 {
347                    future::Either::Left(IdentifyEvent::Received { info, .. }) => {
348                        assert_eq!(info.public_key, pubkey2);
349                        assert_eq!(info.protocol_version, "c");
350                        assert_eq!(info.agent_version, "d");
351                        assert!(!info.protocols.is_empty());
352                        assert!(info.listen_addrs.is_empty());
353                        return;
354                    }
355                    future::Either::Right(IdentifyEvent::Received { info, .. }) => {
356                        assert_eq!(info.public_key, pubkey1);
357                        assert_eq!(info.protocol_version, "a");
358                        assert_eq!(info.agent_version, "b");
359                        assert!(!info.protocols.is_empty());
360                        assert_eq!(info.listen_addrs.len(), 1);
361                        return;
362                    }
363                    _ => {}
364                }
365            }
366        })
367    }
368}
libp2p_identify/identify.rs

libp2p_identify/
identify.rs
