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

urma_node/
peer.rs
