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

urma_node/
peer.rs
