netget/server/libp2p/
host.rs

1//! One libp2p connection, either role: upgrading TCP to Noise and yamux, answering the
2//! peer's identify, identify-push and ping streams in Rust, and turning the application
3//! protocols' streams into messages (uvarint-length-prefixed, the libp2p convention).
4use super::noise::{self, Identity, NoiseConn};
5use super::wire::{self, Io};
6use super::yamux::{Opener, Stream, StreamWriter};
7use super::{bitswap, kad};
8use anyhow::{ensure, Context, Result};
9use std::collections::{BTreeMap, HashMap};
10use std::future::Future;
11use std::net::SocketAddr;
12use std::pin::Pin;
13use std::sync::{Arc, Mutex};
14use std::time::Duration;
15use tokio::net::TcpStream;
16use tokio::sync::mpsc;
17
18/// The application protocol a host speaks when none is configured.
19pub const DEFAULT_PROTOCOLS: &[&str] = &["/netget/chat/1.0.0"];
20/// One application message is at most this long (go-msgio's own default is 8 MiB; NetGet
21/// hands each to a model).
22pub const MAX_MESSAGE: usize = 1024 * 1024;
23/// Negotiating security, the multiplexer, or a stream's protocol.
24pub const NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(10);
25pub const AGENT: &str = concat!("netget/", env!("CARGO_PKG_VERSION"));
26pub const PROTOCOL_VERSION: &str = "ipfs/0.1.0";
27/// Streams whose writers a connection keeps for actions.
28pub const MAX_REMEMBERED_STREAMS: usize = 1024;
29/// A ping is 32 random bytes echoed back.
30pub const PING_SIZE: usize = 32;
31
32pub struct Config {
33    pub identity: Identity,
34    pub protocols: Vec<String>,
35    pub listen_addrs: Vec<Vec<u8>>,
36    /// Accept Bitswap streams (`/ipfs/bitswap/1.2.0` and older) and advertise them.
37    pub bitswap: bool,
38    /// Accept Kademlia DHT streams (`/ipfs/kad/1.0.0`, `/ipfs/lan/kad/1.0.0`) and advertise
39    /// them, which makes this host a DHT server in its peers' eyes.
40    pub kad: bool,
41    /// The peers connected to this host and the addresses their identify gave, for DHT
42    /// answers. One entry per live connection, so it is bounded by the connection limit.
43    pub directory: Directory,
44}
45
46/// Connected peers (base58 id) and their listen addresses (binary multiaddrs).
47pub type Directory = Arc<Mutex<BTreeMap<String, Vec<Vec<u8>>>>>;
48
49impl Config {
50    /// Every protocol this host answers, in the order identify lists them.
51    pub fn served_protocols(&self) -> Vec<String> {
52        let mut out: Vec<String> = [wire::IDENTIFY, wire::IDENTIFY_PUSH, wire::PING]
53            .iter()
54            .map(|s| s.to_string())
55            .collect();
56        if self.bitswap {
57            out.extend(bitswap::PROTOCOLS.iter().map(|s| s.to_string()));
58        }
59        if self.kad {
60            out.extend(kad::PROTOCOLS.iter().map(|s| s.to_string()));
61        }
62        out.extend(self.protocols.iter().cloned());
63        out
64    }
65    /// The longest message a stream of `protocol` may carry.
66    pub fn max_message(&self, protocol: &str) -> usize {
67        if bitswap::is_bitswap(protocol) {
68            bitswap::MAX_MESSAGE
69        } else if kad::is_kad(protocol) {
70            kad::MAX_MESSAGE
71        } else {
72            MAX_MESSAGE
73        }
74    }
75}
76
77/// What a connection's streams report to whoever answers for it.
78pub enum Note {
79    Message {
80        stream_id: u32,
81        protocol: String,
82        data: Vec<u8>,
83    },
84    /// The peer closed its half of a stream (or it failed); `error` says why when it failed.
85    Closed {
86        stream_id: u32,
87        protocol: String,
88        error: Option<String>,
89    },
90    /// The connection ended.
91    Ended(Option<String>),
92}
93
94pub type Task = Pin<Box<dyn Future<Output = ()> + Send>>;
95/// Spawns a task owned by the server or client this connection belongs to.
96pub type Spawn = Arc<dyn Fn(Task) -> Task + Send + Sync>;
97
98/// A live connection: what actions need to open, write and close streams.
99pub struct Peer {
100    pub opener: Opener,
101    pub remote_peer: String,
102    pub remote_addr: SocketAddr,
103    pub streams: Mutex<HashMap<u32, (String, StreamWriter)>>,
104    pub notes: mpsc::Sender<Note>,
105}
106
107impl Peer {
108    pub fn writer(&self, stream_id: u32) -> Option<(String, StreamWriter)> {
109        self.streams
110            .lock()
111            .unwrap_or_else(|e| e.into_inner())
112            .get(&stream_id)
113            .cloned()
114    }
115    /// Keep a stream's writer for later actions; the oldest are dropped past a bound, so a
116    /// peer opening and closing streams forever cannot grow the table.
117    pub fn remember(&self, stream_id: u32, protocol: String, writer: StreamWriter) {
118        let mut t = self.streams.lock().unwrap_or_else(|e| e.into_inner());
119        while t.len() >= MAX_REMEMBERED_STREAMS {
120            let oldest = *t.keys().min().expect("non-empty");
121            t.remove(&oldest);
122        }
123        t.insert(stream_id, (protocol, writer));
124    }
125    pub fn forget(&self, stream_id: u32) {
126        self.streams
127            .lock()
128            .unwrap_or_else(|e| e.into_inner())
129            .remove(&stream_id);
130    }
131}
132
133/// Listener side: select `/noise`, handshake as responder, select `/yamux/1.0.0`.
134pub async fn upgrade_inbound(mut tcp: TcpStream, cfg: &Config) -> Result<NoiseConn> {
135    tokio::time::timeout(NEGOTIATION_TIMEOUT, async {
136        wire::ms_listen(&mut tcp, |p| p == wire::NOISE).await?;
137        let mut conn = noise::handshake(tcp, &cfg.identity, false, None).await?;
138        wire::ms_listen(&mut conn, |p| p == wire::YAMUX).await?;
139        Ok(conn)
140    })
141    .await
142    .context("libp2p upgrade timed out")?
143}
144
145/// Dialler side; `expected` is the peer id the address named, if any.
146pub async fn upgrade_outbound(
147    mut tcp: TcpStream,
148    cfg: &Config,
149    expected: Option<&[u8]>,
150) -> Result<NoiseConn> {
151    tokio::time::timeout(NEGOTIATION_TIMEOUT, async {
152        ensure!(
153            wire::ms_dial(&mut tcp, wire::NOISE).await?,
154            "the peer does not speak /noise"
155        );
156        let mut conn = noise::handshake(tcp, &cfg.identity, true, expected).await?;
157        ensure!(
158            wire::ms_dial(&mut conn, wire::YAMUX).await?,
159            "the peer does not speak /yamux/1.0.0"
160        );
161        Ok(conn)
162    })
163    .await
164    .context("libp2p upgrade timed out")?
165}
166
167/// Serve every stream the peer opens until the connection ends.
168pub async fn accept_streams(
169    mut incoming: mpsc::Receiver<Stream>,
170    cfg: Arc<Config>,
171    peer: Arc<Peer>,
172    spawn: Spawn,
173) {
174    while let Some(stream) = incoming.recv().await {
175        spawn(Box::pin(inbound(stream, cfg.clone(), peer.clone()))).await;
176    }
177}
178
179async fn inbound(mut stream: Stream, cfg: Arc<Config>, peer: Arc<Peer>) {
180    let served = cfg.served_protocols();
181    let supported = |p: &str| served.iter().any(|x| x == p);
182    let writer = stream.writer();
183    let protocol =
184        match tokio::time::timeout(NEGOTIATION_TIMEOUT, wire::ms_listen(&mut stream, supported))
185            .await
186        {
187            Ok(Ok(p)) => p,
188            _ => {
189                writer.reset();
190                return;
191            }
192        };
193    match protocol.as_str() {
194        wire::IDENTIFY => {
195            let msg = wire::encode_identify(&wire::Identify {
196                protocol_version: PROTOCOL_VERSION.into(),
197                agent_version: AGENT.into(),
198                public_key: cfg.identity.public_key_proto.clone(),
199                listen_addrs: cfg.listen_addrs.clone(),
200                observed_addr: Some(wire::multiaddr_bytes(&peer.remote_addr)),
201                protocols: cfg.served_protocols(),
202            });
203            let _ = writer.send(wire::length_prefixed(&msg));
204            writer.close();
205        }
206        wire::IDENTIFY_PUSH => {
207            // Read (bounded) and let it go: NetGet keeps what the first identify said.
208            let _ = wire::read_length_prefixed(&mut stream, wire::MAX_IDENTIFY_BYTES).await;
209            writer.close();
210        }
211        wire::PING => {
212            let mut b = [0u8; PING_SIZE];
213            while stream.read_exact(&mut b).await.is_ok() {
214                if writer.send(b.to_vec()).is_err() {
215                    break;
216                }
217            }
218            writer.close();
219        }
220        _ => {
221            peer.remember(stream.id, protocol.clone(), writer);
222            let max = cfg.max_message(&protocol);
223            read_messages(stream, protocol, peer.notes.clone(), max).await;
224        }
225    }
226}
227
228/// Turn a stream's messages (each at most `max` bytes) into notes until the peer closes it.
229pub async fn read_messages(
230    mut stream: Stream,
231    protocol: String,
232    notes: mpsc::Sender<Note>,
233    max: usize,
234) {
235    let error = loop {
236        match wire::read_length_prefixed(&mut stream, max).await {
237            Ok(data) => {
238                let note = Note::Message {
239                    stream_id: stream.id,
240                    protocol: protocol.clone(),
241                    data,
242                };
243                if notes.send(note).await.is_err() {
244                    return;
245                }
246            }
247            Err(e) => {
248                let text = e.to_string();
249                break (!text.contains("closed by the peer")).then_some(text);
250            }
251        }
252    };
253    if error.is_some() {
254        stream.writer().reset();
255    }
256    let _ = notes
257        .send(Note::Closed {
258            stream_id: stream.id,
259            protocol,
260            error,
261        })
262        .await;
263}
264
265/// What the remote said about itself.
266pub struct Remote {
267    pub agent_version: String,
268    pub protocol_version: String,
269    pub protocols: Vec<String>,
270    pub listen_addrs: Vec<String>,
271    pub observed_addr: Option<String>,
272}
273
274/// Ask the remote's identify.
275pub async fn identify(peer: &Peer) -> Result<Remote> {
276    tokio::time::timeout(NEGOTIATION_TIMEOUT, async {
277        let mut s = peer.opener.open()?;
278        ensure!(
279            wire::ms_dial(&mut s, wire::IDENTIFY).await?,
280            "the peer does not answer identify"
281        );
282        let msg = wire::read_length_prefixed(&mut s, wire::MAX_IDENTIFY_BYTES).await?;
283        s.writer().close();
284        let id = wire::decode_identify(&msg)?;
285        Ok(Remote {
286            agent_version: id.agent_version,
287            protocol_version: id.protocol_version,
288            protocols: id.protocols,
289            listen_addrs: id
290                .listen_addrs
291                .iter()
292                .filter_map(|a| wire::multiaddr_text(a).ok())
293                .collect(),
294            observed_addr: id.observed_addr.and_then(|a| wire::multiaddr_text(&a).ok()),
295        })
296    })
297    .await
298    .context("identify timed out")?
299}
300
301/// Ping the remote once: the round trip.
302pub async fn ping(peer: &Peer) -> Result<Duration> {
303    tokio::time::timeout(NEGOTIATION_TIMEOUT, async {
304        let mut s = peer.opener.open()?;
305        ensure!(
306            wire::ms_dial(&mut s, wire::PING).await?,
307            "the peer does not answer ping"
308        );
309        let payload: [u8; PING_SIZE] = rand::random();
310        let started = crate::utils::clock::Instant::now();
311        s.write_all(&payload).await?;
312        let mut back = [0u8; PING_SIZE];
313        s.read_exact(&mut back).await?;
314        let rtt = started.elapsed();
315        s.writer().close();
316        ensure!(back == payload, "the ping came back different");
317        Ok(rtt)
318    })
319    .await
320    .context("ping timed out")?
321}
322
323/// Open a stream for an application protocol; None when the peer refuses the protocol.
324/// Its messages arrive as notes, read by the task this returns.
325pub async fn open_app_stream(peer: &Arc<Peer>, protocol: &str) -> Result<Option<(u32, Task)>> {
326    let mut s = peer.opener.open()?;
327    let accepted = tokio::time::timeout(NEGOTIATION_TIMEOUT, wire::ms_dial(&mut s, protocol))
328        .await
329        .context("protocol negotiation timed out")??;
330    if !accepted {
331        s.writer().close();
332        return Ok(None);
333    }
334    let id = s.id;
335    peer.remember(id, protocol.to_string(), s.writer());
336    let reader: Task = Box::pin(read_messages(
337        s,
338        protocol.to_string(),
339        peer.notes.clone(),
340        MAX_MESSAGE,
341    ));
342    Ok(Some((id, reader)))
343}
344
345/// Open a stream proposing each of `protocols` in turn: the stream and the protocol the peer
346/// accepted, or None when it accepts none of them. The caller owns the stream.
347pub async fn open_negotiated(peer: &Peer, protocols: &[&str]) -> Result<Option<(Stream, String)>> {
348    let mut s = peer.opener.open()?;
349    let accepted = tokio::time::timeout(NEGOTIATION_TIMEOUT, wire::ms_dial_any(&mut s, protocols))
350        .await
351        .context("protocol negotiation timed out")??;
352    match accepted {
353        Some(p) => Ok(Some((s, p))),
354        None => {
355            s.writer().close();
356            Ok(None)
357        }
358    }
359}
360
361/// Write one message to a stream this connection has open.
362pub fn send(peer: &Peer, stream_id: u32, data: &[u8]) -> Result<()> {
363    ensure!(
364        data.len() <= MAX_MESSAGE,
365        "message longer than {MAX_MESSAGE} bytes"
366    );
367    let (_, w) = peer
368        .writer(stream_id)
369        .with_context(|| format!("no open stream {stream_id}"))?;
370    w.send(wire::length_prefixed(data))
371}
netget/server/libp2p/host.rs

netget/server/libp2p/
host.rs
