network/
discovery.rs

1// Copyright 2019 Conflux Foundation. All rights reserved.
2// Conflux is free software and distributed under GNU General Public License.
3// See http://www.gnu.org/licenses/
4
5use crate::{
6    hash::keccak,
7    node_database::NodeDatabase,
8    node_table::{NodeId, *},
9    service::{UdpIoContext, MAX_DATAGRAM_SIZE, UDP_PROTOCOL_DISCOVERY},
10    DiscoveryConfiguration, Error, IpFilter, ThrottlingReason,
11    NODE_TAG_ARCHIVE, NODE_TAG_NODE_TYPE,
12};
13use cfx_bytes::Bytes;
14use cfx_types::{H256, H520};
15use cfx_util_macros::bail;
16use cfxkey::{recover, sign, KeyPair, Secret};
17use log::{debug, trace, warn};
18use rlp::{Encodable, Rlp, RlpStream};
19use rlp_derive::{RlpDecodable, RlpEncodable};
20use std::{
21    collections::{hash_map::Entry, HashMap, HashSet},
22    net::{IpAddr, SocketAddr},
23    time::{Instant, SystemTime, UNIX_EPOCH},
24};
25use throttling::time_window_bucket::TimeWindowBucket;
26
27const DISCOVER_PROTOCOL_VERSION: u32 = 1;
28
29const DISCOVERY_MAX_STEPS: u16 = 4; // Max iterations of discovery. (discover)
30
31const PACKET_PING: u8 = 1;
32const PACKET_PONG: u8 = 2;
33const PACKET_FIND_NODE: u8 = 3;
34const PACKET_NEIGHBOURS: u8 = 4;
35
36struct PingRequest {
37    // Time when the request was sent
38    sent_at: Instant,
39    // The node to which the request was sent
40    node: NodeEntry,
41    // The hash sent in the Ping request
42    echo_hash: H256,
43}
44
45struct FindNodeRequest {
46    // Time when the request was sent
47    sent_at: Instant,
48    // Number of neighbor chunks for the response
49    num_chunks: usize,
50    // Number of received chunks for the response
51    received_chunks: HashSet<usize>,
52}
53
54impl Default for FindNodeRequest {
55    fn default() -> Self {
56        FindNodeRequest {
57            sent_at: Instant::now(),
58            num_chunks: 0,
59            received_chunks: HashSet::new(),
60        }
61    }
62}
63
64impl FindNodeRequest {
65    fn is_completed(&self) -> bool {
66        self.num_chunks > 0 && self.num_chunks == self.received_chunks.len()
67    }
68}
69
70#[allow(dead_code)]
71pub struct Discovery {
72    id: NodeId,
73    id_hash: H256,
74    secret: Secret,
75    public_endpoint: NodeEndpoint,
76    discovery_initiated: bool,
77    discovery_round: Option<u16>,
78    discovery_nodes: HashSet<NodeId>,
79    in_flight_pings: HashMap<NodeId, PingRequest>,
80    in_flight_find_nodes: HashMap<NodeId, FindNodeRequest>,
81    check_timestamps: bool,
82    adding_nodes: Vec<NodeEntry>,
83    ip_filter: IpFilter,
84    pub disc_option: DiscoveryOption,
85
86    // Limits the response for PING/FIND_NODE packets
87    ping_throttling: TimeWindowBucket<IpAddr>,
88    find_nodes_throttling: TimeWindowBucket<IpAddr>,
89
90    config: DiscoveryConfiguration,
91}
92
93impl Discovery {
94    pub fn new(
95        key: &KeyPair, public: NodeEndpoint, ip_filter: IpFilter,
96        config: DiscoveryConfiguration,
97    ) -> Discovery {
98        Discovery {
99            id: *key.public(),
100            id_hash: keccak(key.public()),
101            secret: key.secret().clone(),
102            public_endpoint: public,
103            discovery_initiated: false,
104            discovery_round: None,
105            discovery_nodes: HashSet::new(),
106            in_flight_pings: HashMap::new(),
107            in_flight_find_nodes: HashMap::new(),
108            check_timestamps: true,
109            adding_nodes: Vec::new(),
110            ip_filter,
111            disc_option: DiscoveryOption {
112                general: true,
113                archive: false,
114            },
115            ping_throttling: TimeWindowBucket::new(
116                config.throttling_interval,
117                config.throttling_limit_ping,
118            ),
119            find_nodes_throttling: TimeWindowBucket::new(
120                config.throttling_interval,
121                config.throttling_limit_find_nodes,
122            ),
123            config,
124        }
125    }
126
127    fn is_allowed(&self, entry: &NodeEntry) -> bool {
128        entry.endpoint.is_allowed(&self.ip_filter) && entry.id != self.id
129    }
130
131    pub fn try_ping_nodes(
132        &mut self, uio: &UdpIoContext, nodes: Vec<NodeEntry>,
133    ) {
134        for node in nodes {
135            self.try_ping(uio, node);
136        }
137    }
138
139    fn try_ping(&mut self, uio: &UdpIoContext, node: NodeEntry) {
140        if !self.is_allowed(&node) {
141            trace!("Node {:?} not allowed", node);
142            return;
143        }
144        if self.in_flight_pings.contains_key(&node.id)
145            || self.in_flight_find_nodes.contains_key(&node.id)
146        {
147            trace!("Node {:?} in flight requests", node);
148            return;
149        }
150        if self.adding_nodes.iter().any(|n| n.id == node.id) {
151            trace!("Node {:?} in adding nodes", node);
152            return;
153        }
154
155        if self.in_flight_pings.len() < self.config.max_nodes_ping {
156            self.ping(uio, &node).unwrap_or_else(|e| {
157                warn!("Error sending Ping packet: {:?}", e);
158            });
159        } else {
160            self.adding_nodes.push(node);
161        }
162    }
163
164    fn ping(
165        &mut self, uio: &UdpIoContext, node: &NodeEntry,
166    ) -> Result<(), Error> {
167        let mut rlp = RlpStream::new_list(4);
168        rlp.append(&DISCOVER_PROTOCOL_VERSION);
169        self.public_endpoint.to_rlp_list(&mut rlp);
170        node.endpoint.to_rlp_list(&mut rlp);
171        rlp.append(&self.config.expire_timestamp());
172        let hash = self.send_packet(
173            uio,
174            PACKET_PING,
175            &node.endpoint.udp_address(),
176            &rlp.out(),
177        )?;
178
179        self.in_flight_pings.insert(
180            node.id,
181            PingRequest {
182                sent_at: Instant::now(),
183                node: node.clone(),
184                echo_hash: hash,
185            },
186        );
187
188        trace!("Sent Ping to {:?} ; node_id={:#x}", &node.endpoint, node.id);
189        Ok(())
190    }
191
192    fn send_packet(
193        &mut self, uio: &UdpIoContext, packet_id: u8, address: &SocketAddr,
194        payload: &[u8],
195    ) -> Result<H256, Error> {
196        let packet = assemble_packet(packet_id, payload, &self.secret)?;
197        let hash = H256::from_slice(&packet[1..=32]);
198        self.send_to(uio, packet, *address);
199        Ok(hash)
200    }
201
202    fn send_to(
203        &mut self, uio: &UdpIoContext, payload: Bytes, address: SocketAddr,
204    ) {
205        uio.send(payload, address);
206    }
207
208    pub fn on_packet(
209        &mut self, uio: &UdpIoContext, packet: &[u8], from: SocketAddr,
210    ) -> Result<(), Error> {
211        // validate packet
212        if packet.len() < 32 + 65 + 4 + 1 {
213            return Err(Error::BadProtocol);
214        }
215
216        let hash_signed = keccak(&packet[32..]);
217        if hash_signed[..] != packet[0..32] {
218            return Err(Error::BadProtocol);
219        }
220
221        let signed = &packet[(32 + 65)..];
222        let signature = H520::from_slice(&packet[32..(32 + 65)]);
223        let node_id = recover(&signature.into(), &keccak(signed))?;
224
225        let packet_id = signed[0];
226        let rlp = Rlp::new(&signed[1..]);
227        match packet_id {
228            PACKET_PING => {
229                self.on_ping(uio, &rlp, &node_id, &from, hash_signed.as_bytes())
230            }
231            PACKET_PONG => self.on_pong(uio, &rlp, &node_id, &from),
232            PACKET_FIND_NODE => self.on_find_node(uio, &rlp, &node_id, &from),
233            PACKET_NEIGHBOURS => self.on_neighbours(uio, &rlp, &node_id, &from),
234            _ => {
235                debug!("Unknown UDP packet: {}", packet_id);
236                Ok(())
237            }
238        }
239    }
240
241    /// Validate that given timestamp is in within one second of now or in the
242    /// future
243    fn check_timestamp(&self, timestamp: u64) -> Result<(), Error> {
244        let secs_since_epoch = SystemTime::now()
245            .duration_since(UNIX_EPOCH)
246            .unwrap_or_default()
247            .as_secs();
248        if self.check_timestamps && timestamp < secs_since_epoch {
249            debug!("Expired packet");
250            return Err(Error::Expired);
251        }
252        Ok(())
253    }
254
255    fn on_ping(
256        &mut self, uio: &UdpIoContext, rlp: &Rlp, node_id: &NodeId,
257        from: &SocketAddr, echo_hash: &[u8],
258    ) -> Result<(), Error> {
259        trace!("Got Ping from {:?}", &from);
260
261        if !self.ping_throttling.try_acquire(from.ip()) {
262            return Err(Error::Throttling(ThrottlingReason::PacketThrottled(
263                "PING",
264            )));
265        }
266
267        let ping_from = NodeEndpoint::from_rlp(&rlp.at(1)?)?;
268        let ping_to = NodeEndpoint::from_rlp(&rlp.at(2)?)?;
269        let timestamp: u64 = rlp.val_at(3)?;
270        self.check_timestamp(timestamp)?;
271
272        let mut response = RlpStream::new_list(3);
273        let pong_to = NodeEndpoint {
274            address: *from,
275            udp_port: ping_from.udp_port,
276        };
277        // Here the PONG's `To` field should be the node we are
278        // sending the request to
279        // WARNING: this field _should not be used_, but old Parity versions
280        // use it in order to get the node's address.
281        // So this is a temporary fix so that older Parity versions don't brake
282        // completely.
283        ping_to.to_rlp_list(&mut response);
284        // pong_to.to_rlp_list(&mut response);
285
286        response.append(&echo_hash);
287        response.append(&self.config.expire_timestamp());
288        self.send_packet(uio, PACKET_PONG, from, &response.out())?;
289
290        let entry = NodeEntry {
291            id: *node_id,
292            endpoint: pong_to,
293        };
294        // TODO handle the error before sending pong
295        if !entry.endpoint.is_valid() {
296            debug!("Got bad address: {:?}", entry);
297        } else if !self.is_allowed(&entry) {
298            debug!("Address not allowed: {:?}", entry);
299        } else {
300            uio.node_db
301                .write()
302                .note_success(node_id, None, false /* trusted_only */);
303        }
304        Ok(())
305    }
306
307    fn on_pong(
308        &mut self, uio: &UdpIoContext, rlp: &Rlp, node_id: &NodeId,
309        from: &SocketAddr,
310    ) -> Result<(), Error> {
311        trace!("Got Pong from {:?} ; node_id={:#x}", &from, node_id);
312        let _pong_to = NodeEndpoint::from_rlp(&rlp.at(0)?)?;
313        let echo_hash: H256 = rlp.val_at(1)?;
314        let timestamp: u64 = rlp.val_at(2)?;
315        self.check_timestamp(timestamp)?;
316
317        let expected_node = match self.in_flight_pings.entry(*node_id) {
318            Entry::Occupied(entry) => {
319                let expected_node = {
320                    let request = entry.get();
321                    if request.echo_hash != echo_hash {
322                        debug!("Got unexpected Pong from {:?} ; packet_hash={:#x} ; expected_hash={:#x}", &from, request.echo_hash, echo_hash);
323                        None
324                    } else {
325                        Some(request.node.clone())
326                    }
327                };
328
329                if expected_node.is_some() {
330                    entry.remove();
331                }
332                expected_node
333            }
334            Entry::Vacant(_) => None,
335        };
336
337        if let Some(node) = expected_node {
338            uio.node_db.write().insert_with_conditional_promotion(node);
339            Ok(())
340        } else {
341            debug!("Got unexpected Pong from {:?} ; request not found", &from);
342            Ok(())
343        }
344    }
345
346    fn on_find_node(
347        &mut self, uio: &UdpIoContext, rlp: &Rlp, _node: &NodeId,
348        from: &SocketAddr,
349    ) -> Result<(), Error> {
350        trace!("Got FindNode from {:?}", &from);
351
352        if !self.find_nodes_throttling.try_acquire(from.ip()) {
353            return Err(Error::Throttling(ThrottlingReason::PacketThrottled(
354                "FIND_NODES",
355            )));
356        }
357
358        let msg: FindNodeMessage = rlp.as_val()?;
359        self.check_timestamp(msg.expire_timestamp)?;
360        let neighbors = msg.sample(
361            &uio.node_db.read(),
362            &self.ip_filter,
363            self.config.discover_node_count,
364        )?;
365
366        trace!("Sample {} Neighbours for {:?}", neighbors.len(), &from);
367
368        let chunk_size = (MAX_DATAGRAM_SIZE - (1 + 109)) / 90;
369        let chunks = NeighborsChunkMessage::chunks(neighbors, chunk_size);
370
371        for chunk in &chunks {
372            self.send_packet(uio, PACKET_NEIGHBOURS, from, &chunk.rlp_bytes())?;
373        }
374
375        trace!("Sent {} Neighbours chunks to {:?}", chunks.len(), &from);
376        Ok(())
377    }
378
379    fn on_neighbours(
380        &mut self, uio: &UdpIoContext, rlp: &Rlp, node_id: &NodeId,
381        from: &SocketAddr,
382    ) -> Result<(), Error> {
383        let mut entry = match self.in_flight_find_nodes.entry(*node_id) {
384            Entry::Occupied(entry) => entry,
385            Entry::Vacant(_) => {
386                debug!("Got unexpected Neighbors from {:?} ; couldn't find node_id={:#x}", &from, node_id);
387                return Ok(());
388            }
389        };
390
391        let msg: NeighborsChunkMessage = rlp.as_val()?;
392        let request = entry.get_mut();
393
394        if !msg.update(request)? {
395            return Ok(());
396        }
397
398        if request.is_completed() {
399            entry.remove();
400        }
401
402        trace!("Got {} Neighbours from {:?}", msg.neighbors.len(), &from);
403
404        for node in msg.neighbors {
405            if !node.endpoint.is_valid() {
406                debug!("Bad address: {:?}", node.endpoint);
407                continue;
408            }
409            if node.id == self.id {
410                continue;
411            }
412            if !self.is_allowed(&node) {
413                debug!("Address not allowed: {:?}", node);
414                continue;
415            }
416            self.try_ping(uio, node);
417        }
418
419        Ok(())
420    }
421
422    /// Starts the discovery process at round 0
423    fn start(&mut self) {
424        trace!("Starting discovery");
425        self.discovery_round = Some(0);
426        self.discovery_nodes.clear();
427    }
428
429    /// Complete the discovery process
430    fn stop(&mut self) {
431        trace!("Completing discovery");
432        self.discovery_round = None;
433        self.discovery_nodes.clear();
434    }
435
436    fn check_expired(&mut self, uio: &UdpIoContext, time: Instant) {
437        let mut nodes_to_expire = Vec::new();
438        let ping_timeout = &self.config.ping_timeout;
439        self.in_flight_pings.retain(|node_id, ping_request| {
440            if time.duration_since(ping_request.sent_at) > *ping_timeout {
441                debug!(
442                    "Removing expired PING request for node_id={:#x}",
443                    node_id
444                );
445                nodes_to_expire.push(*node_id);
446                false
447            } else {
448                true
449            }
450        });
451        let find_node_timeout = &self.config.find_node_timeout;
452        self.in_flight_find_nodes.retain(|node_id, find_node_request| {
453            if time.duration_since(find_node_request.sent_at) > *find_node_timeout {
454                if !find_node_request.is_completed() {
455                    debug!("Removing expired FIND NODE request for node_id={:#x}", node_id);
456                    nodes_to_expire.push(*node_id);
457                }
458                false
459            } else {
460                true
461            }
462        });
463        for node_id in nodes_to_expire {
464            self.expire_node_request(uio, node_id);
465        }
466    }
467
468    fn expire_node_request(&mut self, uio: &UdpIoContext, node_id: NodeId) {
469        uio.node_db.write().note_failure(
470            &node_id, false, /* by_connection */
471            true,  /* trusted_only */
472        );
473    }
474
475    fn update_new_nodes(&mut self, uio: &UdpIoContext) {
476        while self.in_flight_pings.len() < self.config.max_nodes_ping {
477            match self.adding_nodes.pop() {
478                Some(next) => self.try_ping(uio, next),
479                None => break,
480            }
481        }
482    }
483
484    fn discover(&mut self, uio: &UdpIoContext) {
485        let discovery_round = match self.discovery_round {
486            Some(r) => r,
487            None => return,
488        };
489        if discovery_round == DISCOVERY_MAX_STEPS {
490            trace!("Discover stop due to beyond max round count.");
491            self.stop();
492            return;
493        }
494        trace!("Starting round {:?}", self.discovery_round);
495        let mut tried_count = 0;
496
497        if self.disc_option.general {
498            tried_count += self.discover_without_tag(uio);
499        }
500
501        if self.disc_option.archive {
502            let key: String = NODE_TAG_NODE_TYPE.into();
503            let value: String = NODE_TAG_ARCHIVE.into();
504            tried_count += self.discover_with_tag(uio, &key, &value);
505        }
506
507        if tried_count == 0 {
508            trace!("Discovery stop due to 0 tried_count");
509            self.stop();
510            return;
511        }
512        self.discovery_round = Some(discovery_round + 1);
513    }
514
515    fn send_find_node(
516        &mut self, uio: &UdpIoContext, node: &NodeEntry,
517        tag_key: Option<String>, tag_value: Option<String>,
518    ) -> Result<(), Error> {
519        let msg = FindNodeMessage::new(
520            tag_key,
521            tag_value,
522            self.config.expire_timestamp(),
523        );
524
525        self.send_packet(
526            uio,
527            PACKET_FIND_NODE,
528            &node.endpoint.udp_address(),
529            &msg.rlp_bytes(),
530        )?;
531
532        self.in_flight_find_nodes
533            .insert(node.id, FindNodeRequest::default());
534
535        trace!("Sent FindNode to {:?}", node);
536        Ok(())
537    }
538
539    pub fn round(&mut self, uio: &UdpIoContext) {
540        self.check_expired(uio, Instant::now());
541        self.update_new_nodes(uio);
542
543        if self.discovery_round.is_some() {
544            self.discover(uio);
545        } else if self.in_flight_pings.is_empty() && !self.discovery_initiated {
546            // Start discovering if the first pings have been sent (or timed
547            // out)
548            self.discovery_initiated = true;
549            self.refresh();
550        }
551    }
552
553    pub fn refresh(&mut self) {
554        if self.discovery_round.is_none() {
555            self.start();
556        }
557    }
558
559    fn discover_without_tag(&mut self, uio: &UdpIoContext) -> usize {
560        let sampled: Vec<NodeEntry> = uio
561            .node_db
562            .read()
563            .sample_trusted_nodes(
564                self.config.discover_node_count,
565                &self.ip_filter,
566            )
567            .into_iter()
568            .filter(|n| !self.discovery_nodes.contains(&n.id))
569            .collect();
570
571        self.discover_with_nodes(uio, sampled, None, None)
572    }
573
574    fn discover_with_nodes(
575        &mut self, uio: &UdpIoContext, nodes: Vec<NodeEntry>,
576        tag_key: Option<String>, tag_value: Option<String>,
577    ) -> usize {
578        let mut sent = 0;
579
580        for node in nodes {
581            match self.send_find_node(
582                uio,
583                &node,
584                tag_key.clone(),
585                tag_value.clone(),
586            ) {
587                Ok(_) => {
588                    self.discovery_nodes.insert(node.id);
589                    sent += 1;
590                }
591                Err(e) => {
592                    warn!(
593                        "Error sending node discovery packet for {:?}: {:?}",
594                        node.endpoint, e
595                    );
596                }
597            }
598        }
599
600        sent
601    }
602
603    fn discover_with_tag(
604        &mut self, uio: &UdpIoContext, key: &String, value: &String,
605    ) -> usize {
606        let tagged_nodes = uio.node_db.read().sample_trusted_node_ids_with_tag(
607            self.config.discover_node_count / 2,
608            key,
609            value,
610        );
611
612        let count = self.config.discover_node_count - tagged_nodes.len() as u32;
613        let random_nodes = uio
614            .node_db
615            .read()
616            .sample_trusted_node_ids(count, &self.ip_filter);
617
618        let sampled: HashSet<NodeId> = tagged_nodes
619            .into_iter()
620            .chain(random_nodes)
621            .filter(|id| !self.discovery_nodes.contains(id))
622            .collect();
623
624        let sampled_nodes = uio
625            .node_db
626            .read()
627            .get_nodes(sampled, true /* trusted_only */);
628
629        self.discover_with_nodes(
630            uio,
631            sampled_nodes,
632            Some(key.clone()),
633            Some(value.clone()),
634        )
635    }
636}
637
638fn assemble_packet(
639    packet_id: u8, bytes: &[u8], secret: &Secret,
640) -> Result<Bytes, Error> {
641    let mut packet = Bytes::with_capacity(bytes.len() + 32 + 65 + 1 + 1);
642    packet.push(UDP_PROTOCOL_DISCOVERY);
643    packet.resize(1 + 32 + 65, 0); // Filled in below
644    packet.push(packet_id);
645    packet.extend_from_slice(bytes);
646
647    let hash = keccak(&packet[(1 + 32 + 65)..]);
648    let signature = match sign(secret, &hash) {
649        Ok(s) => s,
650        Err(e) => {
651            warn!("Error signing UDP packet");
652            return Err(Error::from(e));
653        }
654    };
655    packet[(1 + 32)..(1 + 32 + 65)].copy_from_slice(&signature[..]);
656    let signed_hash = keccak(&packet[(1 + 32)..]);
657    packet[1..=32].copy_from_slice(signed_hash.as_bytes());
658    Ok(packet)
659}
660
661pub struct DiscoveryOption {
662    // discover nodes without any tag filter
663    pub general: bool,
664    // discover archive nodes
665    pub archive: bool,
666}
667
668#[derive(RlpEncodable, RlpDecodable)]
669struct FindNodeMessage {
670    pub tag_key: Option<String>,
671    pub tag_value: Option<String>,
672    pub expire_timestamp: u64,
673}
674
675impl FindNodeMessage {
676    fn new(
677        tag_key: Option<String>, tag_value: Option<String>,
678        expire_timestamp: u64,
679    ) -> Self {
680        FindNodeMessage {
681            tag_key,
682            tag_value,
683            expire_timestamp,
684        }
685    }
686
687    fn sample(
688        &self, node_db: &NodeDatabase, ip_filter: &IpFilter,
689        discover_node_count: u32,
690    ) -> Result<Vec<NodeEntry>, Error> {
691        let key = match self.tag_key {
692            Some(ref key) => key,
693            None => {
694                return Ok(node_db
695                    .sample_trusted_nodes(discover_node_count, ip_filter))
696            }
697        };
698
699        let value = match self.tag_value {
700            Some(ref value) => value,
701            None => return Err(Error::BadProtocol),
702        };
703
704        let ids = node_db.sample_trusted_node_ids_with_tag(
705            discover_node_count,
706            key,
707            value,
708        );
709
710        Ok(node_db.get_nodes(ids, true /* trusted_onlys */))
711    }
712}
713
714#[derive(RlpEncodable, RlpDecodable)]
715struct NeighborsChunkMessage {
716    neighbors: Vec<NodeEntry>,
717    num_chunks: usize,
718    chunk_index: usize,
719}
720
721impl NeighborsChunkMessage {
722    fn chunks(
723        neighbors: Vec<NodeEntry>, chunk_size: usize,
724    ) -> Vec<NeighborsChunkMessage> {
725        let chunks = neighbors.chunks(chunk_size);
726        let num_chunks = chunks.len();
727        chunks
728            .enumerate()
729            .map(|(chunk_index, chunk)| NeighborsChunkMessage {
730                neighbors: chunk.to_vec(),
731                num_chunks,
732                chunk_index,
733            })
734            .collect()
735    }
736
737    fn validate(&self) -> Result<(), Error> {
738        if self.neighbors.is_empty() {
739            debug!("invalid NeighborsChunkMessage, neighbors is empty");
740            bail!(Error::BadProtocol);
741        }
742
743        if self.num_chunks == 0 {
744            debug!("invalid NeighborsChunkMessage, num_chunks is zero");
745            bail!(Error::BadProtocol);
746        }
747
748        if self.chunk_index >= self.num_chunks {
749            debug!(
750                "invalid NeighborsChunkMessage, chunk index is invalid, len = {}, index = {}",
751                self.num_chunks, self.chunk_index
752            );
753            bail!(Error::BadProtocol);
754        }
755
756        Ok(())
757    }
758
759    /// updates the find node request with this message.
760    /// Return Ok(true) if new chunk received.
761    /// Return Ok(false) if duplicated chunk received.
762    /// Return Err if validation failed.
763    fn update(&self, request: &mut FindNodeRequest) -> Result<bool, Error> {
764        self.validate()?;
765
766        if request.num_chunks == 0 {
767            request.num_chunks = self.num_chunks;
768        } else if request.num_chunks != self.num_chunks {
769            debug!("invalid NeighborsChunkMessage, chunk number mismatch, requested = {}, responded = {}", request.num_chunks, self.num_chunks);
770            bail!(Error::BadProtocol);
771        }
772
773        if !request.received_chunks.insert(self.chunk_index) {
774            debug!("duplicated NeighborsChunkMessage");
775            return Ok(false);
776        }
777
778        Ok(true)
779    }
780}
network/discovery.rs

network/
discovery.rs
