urma_node/
peer.rs

1//! Explicitly started URMA peer transport. Noise authenticates the remote
2//! transport identity; packages still need independent Engine admission.
3//! There are no built-in seeds, wallet keys or implicit listeners. Names in
4//! explicitly dialed `/dns*` addresses resolve through the system resolver.
5use crate::peer_transport::swarm;
6pub(crate) use crate::peer_transport::{Behaviour, BehaviourEvent};
7use crate::{
8    DialLimit, Error,
9    engine::Engine,
10    model::RecordPackage,
11    peer_gossip::{Dial, Dials, Shared, is_loopback, validate_announce, validate_peers},
12    peer_inventory::{lookup, validate_inventory},
13    peer_limits::LimitDenied,
14    peer_live::Link,
15    peer_parent::{Parent, validate_parent},
16    peer_protocol::{GossipPeer, InventoryPage, Kind, Reply, Request, Response, VERSION},
17    peer_rate::{Bucket, Count, peer_rate},
18    peerbook::PeerAddress,
19};
20pub use crate::{
21    peer_live::{PeerSource, SyncPeers},
22    peer_parent::OverlayCommitSource,
23    peer_transport::{Dns, DnsMode},
24};
25use bitcoin::{BlockHash, Txid};
26use libp2p::{
27    Multiaddr, PeerId,
28    futures::StreamExt,
29    identify,
30    identity::Keypair,
31    multiaddr::Protocol,
32    request_response::{self, Message, OutboundRequestId, ResponseChannel},
33    swarm::{ConnectionId, DialError, SwarmEvent},
34};
35use std::{
36    collections::{HashMap, HashSet},
37    sync::{
38        Arc,
39        atomic::{AtomicBool, AtomicU64, Ordering},
40        mpsc,
41    },
42    thread::JoinHandle,
43    time::{Duration, Instant},
44};
45use tokio::sync::mpsc as async_mpsc;
46use urma_chain::observation::Chain;
47
48pub(crate) const REQUEST_TIMEOUT: Duration = Duration::from_secs(8);
49pub(crate) const CALL_TIMEOUT: Duration = Duration::from_secs(10);
50const MAX_PENDING: usize = 8;
51
52/// Owns one network thread and one bounded archive-reader thread. Drop requests
53/// shutdown and joins both. An archive operation already running is not forcibly
54/// interrupted. Source handles do not keep the network alive after this owner.
55pub struct PeerNetwork {
56    peer: PeerId,
57    listen: Option<Multiaddr>,
58    dns: Dns,
59    pub(crate) commands: async_mpsc::Sender<Command>,
60    pub(crate) stop: Arc<AtomicBool>,
61    connected: Arc<AtomicU64>,
62    pub(crate) shared: Arc<Shared>,
63    worker: Option<JoinHandle<Result<(), Error>>>,
64}
65
66pub(crate) enum Command {
67    Fetch(Fetch),
68    /// Fire and forget: frees the slot held by a peer whose reply was rejected.
69    Disconnect(PeerId),
70    Dial(Dial),
71}
72
73/// A request to one peer, over its existing connection or the given addresses.
74pub(crate) struct Fetch {
75    peer: PeerId,
76    addresses: Vec<Multiaddr>,
77    kind: Kind,
78    deadline: Instant,
79    answer: Answer,
80}
81
82pub(crate) struct Pending {
83    peer: PeerId,
84    pub(crate) kind: Kind,
85    deadline: Instant,
86    answer: Answer,
87}
88
89/// Reply channel typed by request kind; a cross-kind reply is a peer error.
90pub(crate) enum Answer {
91    Record(mpsc::SyncSender<Result<Option<RecordPackage>, Error>>),
92    Parent(mpsc::SyncSender<Result<Parent, Error>>),
93    Inventory(mpsc::SyncSender<Result<Option<InventoryPage>, Error>>),
94    Peers(mpsc::SyncSender<Result<Vec<GossipPeer>, Error>>),
95}
96
97impl Answer {
98    fn fail(&self, error: Error) {
99        match self {
100            Answer::Record(sender) => {
101                let _ = sender.try_send(Err(error));
102            }
103            Answer::Parent(sender) => {
104                let _ = sender.try_send(Err(error));
105            }
106            Answer::Inventory(sender) => {
107                let _ = sender.try_send(Err(error));
108            }
109            Answer::Peers(sender) => {
110                let _ = sender.try_send(Err(error));
111            }
112        }
113    }
114
115    fn deliver(&self, genesis: BlockHash, peer: PeerId, pending: &Pending, response: Response) {
116        match self {
117            Answer::Record(sender) => {
118                let _ = sender.try_send(validate_response(genesis, peer, pending, response));
119            }
120            Answer::Parent(sender) => {
121                let _ = sender.try_send(validate_parent(genesis, peer, pending, response));
122            }
123            Answer::Inventory(sender) => {
124                let _ = sender.try_send(validate_inventory(genesis, peer, pending, response));
125            }
126            Answer::Peers(sender) => {
127                let _ = sender.try_send(validate_peers(genesis, peer, pending, response));
128            }
129        }
130    }
131}
132
133type Startup = mpsc::SyncSender<Result<(Option<Multiaddr>, Dns), Error>>;
134struct ServeJob {
135    deadline: Instant,
136    request: Request,
137    channel: ResponseChannel<Response>,
138}
139struct Served {
140    response: Response,
141    channel: ResponseChannel<Response>,
142    bytes: usize,
143}
144
145impl PeerNetwork {
146    /// Listen only when an explicit IP/TCP address is supplied. Port zero is
147    /// allowed for an ephemeral listener; returned address includes this PeerId.
148    /// Merely constructing source handles never dials. No peer is a seed by default.
149    /// `announce` are operator-chosen public IP/TCP addresses identify advertises.
150    pub fn start(
151        engine: Arc<Engine>,
152        identity: Keypair,
153        listen: Option<Multiaddr>,
154        announce: Vec<Multiaddr>,
155    ) -> Result<Self, Error> {
156        Self::start_with_dns(engine, identity, (listen, announce), DnsMode::System)
157    }
158
159    /// As `start`, with the name resolution mode explicit; `DnsMode::Refuse`
160    /// stands in for an unreadable system resolver configuration.
161    pub fn start_with_dns(
162        engine: Arc<Engine>,
163        identity: Keypair,
164        (listen, announce): (Option<Multiaddr>, Vec<Multiaddr>),
165        dns: DnsMode,
166    ) -> Result<Self, Error> {
167        if let Some(address) = &listen {
168            validate_listener(address)?;
169        }
170        engine.status()?;
171        let peer = identity.public().to_peer_id();
172        let announce = announce
173            .iter()
174            .map(|address| validate_announce(address, peer))
175            .collect::<Result<Vec<_>, _>>()?;
176        let (commands, receiver) = async_mpsc::channel(MAX_PENDING);
177        let (startup, ready) = mpsc::sync_channel(1);
178        let stop = Arc::new(AtomicBool::new(false));
179        let connected = Arc::new(AtomicU64::new(0));
180        let shared = Arc::new(Shared::new(peer));
181        let control = Control {
182            commands: receiver,
183            stop: stop.clone(),
184            connected: connected.clone(),
185            startup,
186            shared: shared.clone(),
187            announce,
188        };
189        let worker = std::thread::Builder::new()
190            .name("urma-peer".into())
191            .spawn(move || {
192                let runtime = tokio::runtime::Builder::new_current_thread()
193                    .enable_all()
194                    .build()?;
195                runtime.block_on(run(engine, identity, (listen, dns), control))
196            })?;
197        let (actual, dns) = match ready.recv_timeout(Duration::from_secs(5)) {
198            Ok(Ok(started)) => started,
199            result => {
200                stop.store(true, Ordering::Release);
201                let joined = worker.join();
202                return Err(match result {
203                    Ok(Err(error)) => error,
204                    _ => match joined {
205                        Ok(Err(error)) => error,
206                        _ => Error::Busy("peer listener startup failed or expired".into()),
207                    },
208                });
209            }
210        };
211        Ok(Self {
212            peer,
213            listen: actual,
214            dns,
215            commands,
216            stop,
217            connected,
218            shared,
219            worker: Some(worker),
220        })
221    }
222
223    pub fn peer_id(&self) -> PeerId {
224        self.peer
225    }
226    /// Distinct peers with at least one established connection right now.
227    pub fn connected_peers(&self) -> u64 {
228        self.connected.load(Ordering::Relaxed)
229    }
230    pub fn listen_address(&self) -> Option<Multiaddr> {
231        self.listen.clone()
232    }
233
234    /// Whether named addresses can be dialed by this overlay.
235    pub fn dns(&self) -> &Dns {
236        &self.dns
237    }
238
239    /// Every address the listener is bound to now, with `/p2p`.
240    pub fn listen_addresses(&self) -> Vec<Multiaddr> {
241        self.shared.listeners()
242    }
243
244    fn link(&self) -> Link {
245        Link {
246            commands: self.commands.clone(),
247            stop: self.stop.clone(),
248            shared: Arc::clone(&self.shared),
249        }
250    }
251
252    pub fn source(&self, address: PeerAddress) -> PeerSource {
253        self.link()
254            .source(address.peer_id(), vec![address.address().clone()])
255    }
256
257    /// `manual` peers first, then whoever is connected and on our chain.
258    pub fn sync_peers(&self, manual: Vec<PeerAddress>) -> SyncPeers {
259        SyncPeers::new(self.link(), manual)
260    }
261
262    pub fn commit_source(&self, chain: Chain, manual: Vec<PeerAddress>) -> OverlayCommitSource {
263        OverlayCommitSource::new(chain, self.sync_peers(manual))
264    }
265
266    pub fn shutdown(mut self) -> Result<(), Error> {
267        self.stop.store(true, Ordering::Release);
268        match self.worker.take().expect("network owner has worker").join() {
269            Ok(result) => result,
270            Err(_) => Err(Error::Invalid("peer worker panicked".into())),
271        }
272    }
273}
274
275impl Drop for PeerNetwork {
276    fn drop(&mut self) {
277        self.stop.store(true, Ordering::Release);
278        if let Some(worker) = self.worker.take() {
279            let _ = worker.join();
280        }
281    }
282}
283
284/// One bounded request: CALL_TIMEOUT and the caller's own deadline, if earlier.
285/// Without addresses the peer must already be connected or known to the swarm.
286pub(crate) fn request<T>(
287    commands: &async_mpsc::Sender<Command>,
288    stop: &AtomicBool,
289    (peer, addresses): (PeerId, Vec<Multiaddr>),
290    kind: Kind,
291    limit: Instant,
292    answer: fn(mpsc::SyncSender<Result<T, Error>>) -> Answer,
293) -> Result<T, Error> {
294    if stop.load(Ordering::Acquire) {
295        return Err(Error::Busy("peer network stopped".into()));
296    }
297    let deadline = limit.min(Instant::now() + CALL_TIMEOUT);
298    let (sender, receiver) = mpsc::sync_channel(1);
299    commands
300        .try_send(Command::Fetch(Fetch {
301            peer,
302            addresses,
303            kind,
304            deadline,
305            answer: answer(sender),
306        }))
307        .map_err(|_| Error::Busy("peer command queue unavailable".into()))?;
308    receiver
309        .recv_timeout(deadline.saturating_duration_since(Instant::now()))
310        .map_err(|_| Error::Busy("peer acquisition deadline or shutdown".into()))?
311}
312
313impl PeerSource {
314    pub(crate) fn parent(&self, txid: Txid, deadline: Instant) -> Result<Parent, Error> {
315        self.request(Kind::Parent { txid }, deadline, Answer::Parent)
316    }
317}
318
319fn validate_listener(address: &Multiaddr) -> Result<(), Error> {
320    let mut parts = address.iter();
321    let ip = matches!(
322        parts.next(),
323        Some(Protocol::Ip4(_)) | Some(Protocol::Ip6(_))
324    );
325    if !ip || !matches!(parts.next(), Some(Protocol::Tcp(_))) || parts.next().is_some() {
326        return Err(Error::Invalid(
327            "peer listener requires only IP and TCP".into(),
328        ));
329    }
330    Ok(())
331}
332
333struct Control {
334    commands: async_mpsc::Receiver<Command>,
335    stop: Arc<AtomicBool>,
336    connected: Arc<AtomicU64>,
337    startup: Startup,
338    shared: Arc<Shared>,
339    announce: Vec<Multiaddr>,
340}
341
342/// Remote address per live connection, so gossip can tell loopback requesters.
343#[derive(Default)]
344struct Remotes(HashMap<ConnectionId, (Multiaddr, bool)>);
345
346impl Remotes {
347    fn loopback(&self, connection: ConnectionId) -> bool {
348        self.0
349            .get(&connection)
350            .is_some_and(|(address, _)| is_loopback(address))
351    }
352
353    fn outbound(&self, connection: ConnectionId) -> bool {
354        self.0
355            .get(&connection)
356            .is_some_and(|(_, outbound)| *outbound)
357    }
358}
359
360async fn run(
361    engine: Arc<Engine>,
362    identity: Keypair,
363    (listen, dns): (Option<Multiaddr>, DnsMode),
364    control: Control,
365) -> Result<(), Error> {
366    let Control {
367        mut commands,
368        stop,
369        connected,
370        startup,
371        shared,
372        announce,
373    } = control;
374    let genesis = engine.status()?.chain_genesis;
375    let (mut swarm, dns) = swarm(identity, &dns)?;
376    for address in announce {
377        swarm.add_external_address(address);
378    }
379    let mut ready = Some(startup);
380    let listening = listen.is_some();
381    if let Some(address) = listen {
382        swarm
383            .listen_on(address)
384            .map_err(|e| Error::Invalid(format!("peer listen: {e}")))?;
385    }
386    let (jobs, input) = mpsc::sync_channel::<ServeJob>(2);
387    let (served, mut responses) = async_mpsc::channel::<Served>(2);
388    let serving_stop = stop.clone();
389    let serving = std::thread::Builder::new()
390        .name("urma-peer-read".into())
391        .spawn(move || serve(&engine, genesis, &serving_stop, input, served))?;
392    if !listening {
393        let _ = ready.take().unwrap().send(Ok((None, dns.clone())));
394    }
395    let mut pending = HashMap::<OutboundRequestId, Pending>::new();
396    let mut inbound = HashSet::new();
397    let mut remotes = Remotes::default();
398    let mut dials = Dials::default();
399    let mut rates = HashMap::<PeerId, Bucket>::new();
400    let mut requests = Bucket::new(16, 16);
401    let mut bytes = Bucket::new(64 * 1024 * 1024, 4 * 1024 * 1024);
402    let mut tick = tokio::time::interval(Duration::from_millis(50));
403    let started = Instant::now();
404    let result = loop {
405        if stop.load(Ordering::Acquire) {
406            break Ok(());
407        }
408        if ready.is_some() && started.elapsed() > Duration::from_secs(4) {
409            break Err(Error::Busy("peer listener startup deadline".into()));
410        }
411        tokio::select! {
412            _ = tick.tick() => {
413                // Keep the request tracked until libp2p reports completion. This
414                // bounds its internal pending queue even if a caller times out.
415                for value in pending.values_mut() {
416                    if Instant::now() >= value.deadline {
417                        value.answer.fail(Error::Busy("peer acquisition deadline".into()));
418                        let _ = swarm.disconnect_peer_id(value.peer);
419                    }
420                }
421                dials.expire();
422            }
423            command = commands.recv() => {
424                let command = match command {
425                    Some(Command::Fetch(fetch)) => fetch,
426                    // A Fetch queued for this peer since the rejection keeps its connection.
427                    Some(Command::Disconnect(peer)) => {
428                        if !pending.values().any(|p| p.peer == peer) { let _ = swarm.disconnect_peer_id(peer); }
429                        continue;
430                    }
431                    Some(Command::Dial(dial)) => { dials.start(&mut swarm, dial); continue; }
432                    None => break Ok(()),
433                };
434                if Instant::now() >= command.deadline || pending.len() >= MAX_PENDING
435                    || pending.values().any(|p| p.peer == command.peer) {
436                    command.answer.fail(Error::Busy("peer acquisition capacity or deadline".into()));
437                    continue;
438                }
439                let request = Request { version: VERSION, chain_genesis: genesis, kind: command.kind };
440                let id = swarm.behaviour_mut().records.send_request_with_addresses(
441                    &command.peer, request, command.addresses);
442                pending.insert(id, Pending { peer: command.peer, kind: command.kind, deadline: command.deadline, answer: command.answer });
443            }
444            response = responses.recv() => {
445                let Some(mut response) = response else { break Err(Error::Invalid("peer archive reader stopped".into())); };
446                if response.channel.is_open() {
447                    if !bytes.take(u64::try_from(response.bytes.saturating_sub(256)).unwrap_or(u64::MAX)) { response.response.result = Reply::Busy; }
448                    let _ = swarm.behaviour_mut().records.send_response(response.channel, response.response);
449                }
450            }
451            event = swarm.select_next_some() => match event {
452                SwarmEvent::NewListenAddr { mut address, .. } => {
453                    address.push(Protocol::P2p(*swarm.local_peer_id()));
454                    shared.listening(address.clone(), true);
455                    if let Some(ready) = ready.take() { let _ = ready.send(Ok((Some(address), dns.clone()))); }
456                }
457                SwarmEvent::ExpiredListenAddr { address, .. } => shared.listening(address.with(Protocol::P2p(*swarm.local_peer_id())), false),
458                SwarmEvent::ListenerClosed { addresses, .. } => for address in addresses { shared.listening(address.with(Protocol::P2p(*swarm.local_peer_id())), false) },
459                SwarmEvent::ListenerError { error, .. } if ready.is_some() => break Err(Error::Io(error)),
460                SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, num_established, .. } => {
461                    remotes.0.insert(connection_id, (endpoint.get_remote_address().clone(), endpoint.is_dialer()));
462                    dials.established(connection_id, peer_id);
463                    shared.opened(peer_id, endpoint.is_dialer());
464                    if num_established.get() == 1 {
465                        connected.fetch_add(1, Ordering::Relaxed);
466                    }
467                }
468                SwarmEvent::ConnectionClosed { peer_id, connection_id, endpoint, num_established, .. } => {
469                    remotes.0.remove(&connection_id);
470                    shared.closed(peer_id, endpoint.is_dialer());
471                    if num_established == 0 {
472                        connected.fetch_sub(1, Ordering::Relaxed);
473                    }
474                }
475                SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
476                    // Queued before request-response reports the same DialFailure.
477                    if let (Some(peer), DialError::Denied { cause }) = (peer_id, &error) {
478                        if let Some(LimitDenied(limit)) = cause.downcast_ref::<LimitDenied>() { release_denied(&mut pending, peer, *limit); }
479                    }
480                    dials.failed(connection_id, &error);
481                }
482                SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { connection_id, peer_id, info })) => {
483                    // Inbound peers could steer our address; only dialed peers witness it.
484                    if remotes.outbound(connection_id) {
485                        if let Some(confirmed) = shared.observe(info.observed_addr, peer_id) { swarm.add_external_address(confirmed); }
486                    }
487                    shared.identified(peer_id, info.listen_addrs);
488                }
489                SwarmEvent::Behaviour(BehaviourEvent::Records(event)) => match event {
490                    request_response::Event::Message { peer, connection_id, message: Message::Request { request_id, request, channel } } => {
491                        // Every request, including refusals, consumes rate and
492                        // response allowance. Over-rate requests get an explicit
493                        // Busy; only an exhausted byte allowance drops the channel.
494                        if request.version == VERSION && request.chain_genesis == genesis { shared.on_chain(peer); }
495                        let over_rate = !requests.take(1) || !peer_rate(&mut rates, peer);
496                        if !bytes.take(256) {
497                            continue;
498                        }
499                        let refusal = if over_rate { Some(Reply::Busy) }
500                            else if request.version != VERSION { Some(Reply::UnsupportedVersion) }
501                            else if request.chain_genesis != genesis { Some(Reply::WrongNetwork) }
502                            else if let Kind::Peers { limit } = request.kind {
503                                // Gossip is answered from memory, never by the archive reader.
504                                Some(shared.serve(peer, remotes.loopback(connection_id), limit))
505                            }
506                            else if inbound.len() >= 2 || inbound.iter().any(|(p, _)| *p == peer)
507                                { Some(Reply::Busy) }
508                            else { None };
509                        if let Some(result) = refusal {
510                            let _ = swarm.behaviour_mut().records.send_response(channel, Response { version: VERSION, chain_genesis: genesis, kind: request.kind, result });
511                        } else {
512                            match jobs.try_send(ServeJob { deadline: Instant::now() + REQUEST_TIMEOUT, request, channel }) {
513                                Ok(()) => { inbound.insert((peer, request_id)); }
514                                Err(_) => { /* Dropped response channel refuses the request. */ }
515                            }
516                        }
517                    }
518                    request_response::Event::Message { peer, message: Message::Response { request_id, response }, .. } => {
519                        if response.version == VERSION && response.chain_genesis == genesis { shared.on_chain(peer); }
520                        if let Some(value) = pending.remove(&request_id) {
521                            value.answer.deliver(genesis, peer, &value, response);
522                        }
523                    }
524                    request_response::Event::OutboundFailure { request_id, error, .. } => {
525                        if let Some(value) = pending.remove(&request_id) {
526                            value.answer.fail(Error::Transport(error.to_string()));
527                        }
528                    }
529                    request_response::Event::InboundFailure { peer, request_id, .. }
530                    | request_response::Event::ResponseSent { peer, request_id, .. } => { inbound.remove(&(peer, request_id)); }
531                },
532                _ => (),
533            }
534        }
535    };
536    stop.store(true, Ordering::Release);
537    for (_, value) in pending {
538        value
539            .answer
540            .fail(Error::Busy("peer network stopped".into()));
541    }
542    dials.close();
543    // Drop channels before joining: the archive reader must not wait for a
544    // consumer which is already shutting down. No socket belongs to that reader.
545    drop(swarm);
546    drop(jobs);
547    drop(responses);
548    if serving.join().is_err() {
549        return Err(Error::Invalid("peer archive reader panicked".into()));
550    }
551    result
552}
553
554/// The bounded archive reader: one local lookup per job, sized before sending.
555fn serve(
556    engine: &Engine,
557    genesis: BlockHash,
558    stop: &AtomicBool,
559    input: mpsc::Receiver<ServeJob>,
560    served: async_mpsc::Sender<Served>,
561) {
562    while let Ok(job) = input.recv() {
563        if stop.load(Ordering::Acquire) {
564            break;
565        }
566        if !job.channel.is_open() || Instant::now() >= job.deadline {
567            continue;
568        }
569        let result = lookup(engine, &job.request);
570        if stop.load(Ordering::Acquire) || !job.channel.is_open() || Instant::now() >= job.deadline
571        {
572            continue;
573        }
574        let mut response = Response {
575            version: VERSION,
576            chain_genesis: genesis,
577            kind: job.request.kind,
578            result,
579        };
580        let mut count = Count(0);
581        if serde_json::to_writer(&mut count, &response).is_err() {
582            response.result = Reply::TooLarge;
583            count.0 = 256;
584        }
585        let reply = Served {
586            channel: job.channel,
587            response,
588            bytes: count.0,
589        };
590        if served.blocking_send(reply).is_err() {
591            break;
592        }
593    }
594}
595
596/// A dial refused by our own limits is local capacity: the waiting requests
597/// fail with which limit refused it, never blaming the peer.
598fn release_denied(
599    pending: &mut HashMap<OutboundRequestId, Pending>,
600    peer: PeerId,
601    limit: DialLimit,
602) {
603    pending.retain(|_, value| {
604        if value.peer != peer {
605            return true;
606        }
607        value.answer.fail(Error::DialDenied(limit));
608        false
609    });
610}
611
612pub(crate) fn bind_response(
613    genesis: BlockHash,
614    peer: PeerId,
615    pending: &Pending,
616    response: &Response,
617) -> Result<(), Error> {
618    if Instant::now() >= pending.deadline {
619        return Err(Error::Busy("peer acquisition deadline".into()));
620    }
621    if peer != pending.peer
622        || response.version != VERSION
623        || response.chain_genesis != genesis
624        || response.kind != pending.kind
625    {
626        return Err(Error::Invalid(
627            "peer response identity, network or version mismatch".into(),
628        ));
629    }
630    Ok(())
631}
632
633fn validate_response(
634    genesis: BlockHash,
635    peer: PeerId,
636    pending: &Pending,
637    response: Response,
638) -> Result<Option<RecordPackage>, Error> {
639    bind_response(genesis, peer, pending, &response)?;
640    match response.result {
641        Reply::Record(package) if package.chain_genesis == genesis => Ok(Some(*package)),
642        Reply::Missing => Ok(None),
643        Reply::Busy => Err(Error::PeerBusy),
644        _ => Err(Error::Invalid(
645            "peer refused record or returned wrong network".into(),
646        )),
647    }
648}
649
650#[cfg(test)]
651mod tests {
652    use super::*;
653    use bitcoin::hashes::Hash;
654
655    #[test]
656    fn public_response_metadata_is_bound_but_is_not_record_verification() {
657        let genesis = BlockHash::all_zeros();
658        let txid = Txid::all_zeros();
659        let peer = Keypair::generate_ed25519().public().to_peer_id();
660        let other = Keypair::generate_ed25519().public().to_peer_id();
661        let (answer, _receiver) = mpsc::sync_channel(1);
662        let pending = Pending {
663            peer,
664            kind: Kind::Record { txid },
665            deadline: Instant::now() + CALL_TIMEOUT,
666            answer: Answer::Record(answer),
667        };
668        let response = Response {
669            version: VERSION,
670            chain_genesis: genesis,
671            kind: Kind::Record { txid },
672            result: Reply::Missing,
673        };
674        assert!(
675            validate_response(genesis, peer, &pending, response.clone())
676                .unwrap()
677                .is_none()
678        );
679        assert!(validate_response(genesis, other, &pending, response.clone()).is_err());
680        let mut bad = response.clone();
681        bad.version += 1;
682        assert!(validate_response(genesis, peer, &pending, bad).is_err());
683        let mut bad = response.clone();
684        bad.chain_genesis = BlockHash::from_byte_array([1; 32]);
685        assert!(validate_response(genesis, peer, &pending, bad).is_err());
686        let mut bad = response.clone();
687        bad.kind = Kind::Record {
688            txid: Txid::from_byte_array([1; 32]),
689        };
690        assert!(validate_response(genesis, peer, &pending, bad).is_err());
691        let expired = Pending {
692            deadline: Instant::now(),
693            ..pending
694        };
695        assert!(matches!(
696            validate_response(genesis, peer, &expired, response),
697            Err(Error::Busy(_))
698        ));
699    }
700
701    #[test]
702    fn listener_accepts_only_explicit_ip_tcp_without_implicit_resolution() {
703        for value in [
704            "/ip4/127.0.0.1/tcp/0",
705            "/ip6/::1/tcp/0",
706            "/ip4/0.0.0.0/tcp/28332",
707        ] {
708            validate_listener(&value.parse().unwrap()).unwrap();
709        }
710        for value in [
711            "/dns4/localhost/tcp/1",
712            "/ip4/127.0.0.1/udp/1",
713            "/ip4/127.0.0.1/tcp/1/ws",
714        ] {
715            assert!(validate_listener(&value.parse().unwrap()).is_err());
716        }
717    }
718}
urma_node/peer.rs

urma_node/
peer.rs
