netget/server/libp2p/
wire.rs

1//! The small libp2p encodings, shared by both roles: unsigned varints, the protobuf fields
2//! libp2p uses, base58btc peer ids, binary multiaddrs, multistream-select 1.0 and the
3//! identify message. Each decoder bounds what the peer may announce.
4use anyhow::{bail, ensure, Context, Result};
5use std::future::Future;
6use std::net::{IpAddr, SocketAddr};
7
8/// A multistream-select message (a protocol id and its newline) is at most this long.
9pub const MAX_MULTISTREAM_MESSAGE: usize = 1024;
10/// Proposals a listener answers on one stream before giving up on it.
11pub const MAX_PROPOSALS: usize = 16;
12/// An identify message is at most this long (go-libp2p's own limit is 8 KiB of protobuf
13/// plus the signed peer record; leave room for both).
14pub const MAX_IDENTIFY_BYTES: usize = 64 * 1024;
15pub const MULTISTREAM: &str = "/multistream/1.0.0";
16pub const NOISE: &str = "/noise";
17pub const YAMUX: &str = "/yamux/1.0.0";
18pub const IDENTIFY: &str = "/ipfs/id/1.0.0";
19pub const IDENTIFY_PUSH: &str = "/ipfs/id/push/1.0.0";
20pub const PING: &str = "/ipfs/ping/1.0.0";
21
22/// Byte-stream I/O that every libp2p layer offers the one above it.
23pub trait Io: Send {
24    fn read_exact<'a>(
25        &'a mut self,
26        buf: &'a mut [u8],
27    ) -> impl Future<Output = Result<()>> + Send + 'a;
28    fn write_all<'a>(&'a mut self, buf: &'a [u8]) -> impl Future<Output = Result<()>> + Send + 'a;
29}
30
31impl Io for tokio::net::TcpStream {
32    async fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
33        tokio::io::AsyncReadExt::read_exact(self, buf).await?;
34        Ok(())
35    }
36    async fn write_all(&mut self, buf: &[u8]) -> Result<()> {
37        tokio::io::AsyncWriteExt::write_all(self, buf).await?;
38        Ok(())
39    }
40}
41
42pub fn put_uvarint(out: &mut Vec<u8>, mut v: u64) {
43    while v >= 0x80 {
44        out.push((v as u8) | 0x80);
45        v >>= 7;
46    }
47    out.push(v as u8);
48}
49
50/// Decode a uvarint at the start of `b`: (value, bytes used).
51pub fn get_uvarint(b: &[u8]) -> Result<(u64, usize)> {
52    let mut v = 0u64;
53    for (i, byte) in b.iter().enumerate().take(10) {
54        v |= u64::from(byte & 0x7f) << (7 * i);
55        if byte & 0x80 == 0 {
56            return Ok((v, i + 1));
57        }
58    }
59    bail!("truncated or overlong varint")
60}
61
62/// Read one uvarint from a stream (at most 9 bytes: 63 bits).
63pub async fn read_uvarint(io: &mut impl Io) -> Result<u64> {
64    let mut v = 0u64;
65    for i in 0..9 {
66        let mut b = [0u8];
67        io.read_exact(&mut b).await?;
68        v |= u64::from(b[0] & 0x7f) << (7 * i);
69        if b[0] & 0x80 == 0 {
70            return Ok(v);
71        }
72    }
73    bail!("varint longer than 9 bytes")
74}
75
76/// A uvarint-length-prefixed message, refused before reading past `max`.
77pub async fn read_length_prefixed(io: &mut impl Io, max: usize) -> Result<Vec<u8>> {
78    let n = read_uvarint(io).await?;
79    ensure!(
80        n as usize <= max,
81        "message announces {n} bytes; the limit is {max}"
82    );
83    let mut buf = vec![0u8; n as usize];
84    io.read_exact(&mut buf).await?;
85    Ok(buf)
86}
87
88pub fn length_prefixed(payload: &[u8]) -> Vec<u8> {
89    let mut out = Vec::with_capacity(payload.len() + 5);
90    put_uvarint(&mut out, payload.len() as u64);
91    out.extend_from_slice(payload);
92    out
93}
94
95// ---------------------------------------------------------------------------------------
96// Protobuf: the handful of wire types libp2p's small messages use.
97
98pub fn pb_bytes(out: &mut Vec<u8>, field: u32, value: &[u8]) {
99    put_uvarint(out, u64::from(field << 3 | 2));
100    put_uvarint(out, value.len() as u64);
101    out.extend_from_slice(value);
102}
103
104pub fn pb_varint(out: &mut Vec<u8>, field: u32, value: u64) {
105    put_uvarint(out, u64::from(field << 3));
106    put_uvarint(out, value);
107}
108
109pub enum PbValue<'a> {
110    Varint(u64),
111    Bytes(&'a [u8]),
112}
113
114/// Every (field, value) in a message; unknown wire types are an error, fixed-width ones are
115/// skipped.
116pub fn pb_fields(b: &[u8]) -> Result<Vec<(u32, PbValue<'_>)>> {
117    pb_fields_bounded(b, usize::MAX)
118}
119
120/// [`pb_fields`], refusing a message with more than `max` fields before collecting them, so
121/// a peer cannot turn a few megabytes of two-byte fields into a vector many times larger.
122pub fn pb_fields_bounded(mut b: &[u8], max: usize) -> Result<Vec<(u32, PbValue<'_>)>> {
123    let mut out = Vec::new();
124    while !b.is_empty() {
125        ensure!(
126            out.len() < max,
127            "protobuf message has more than {max} fields"
128        );
129        let (key, n) = get_uvarint(b)?;
130        b = &b[n..];
131        let field = (key >> 3) as u32;
132        match key & 7 {
133            0 => {
134                let (v, n) = get_uvarint(b)?;
135                b = &b[n..];
136                out.push((field, PbValue::Varint(v)));
137            }
138            2 => {
139                let (len, n) = get_uvarint(b)?;
140                b = &b[n..];
141                ensure!(
142                    len as usize <= b.len(),
143                    "protobuf field runs past the message"
144                );
145                out.push((field, PbValue::Bytes(&b[..len as usize])));
146                b = &b[len as usize..];
147            }
148            1 => {
149                ensure!(b.len() >= 8, "truncated fixed64");
150                b = &b[8..];
151            }
152            5 => {
153                ensure!(b.len() >= 4, "truncated fixed32");
154                b = &b[4..];
155            }
156            w => bail!("unsupported protobuf wire type {w}"),
157        }
158    }
159    Ok(out)
160}
161
162// ---------------------------------------------------------------------------------------
163// Keys and peer ids.
164
165/// `PublicKey{Type: Ed25519, Data: key}`, the form a peer id hashes.
166pub fn ed25519_public_key_proto(key: &[u8; 32]) -> Vec<u8> {
167    let mut out = Vec::new();
168    pb_varint(&mut out, 1, 1);
169    pb_bytes(&mut out, 2, key);
170    out
171}
172
173/// The Ed25519 key inside a `PublicKey` protobuf. Other key types are refused.
174pub fn parse_public_key(proto: &[u8]) -> Result<[u8; 32]> {
175    let mut kind = None;
176    let mut data = None;
177    for (f, v) in pb_fields(proto)? {
178        match (f, v) {
179            (1, PbValue::Varint(k)) => kind = Some(k),
180            (2, PbValue::Bytes(d)) => data = Some(d),
181            _ => {}
182        }
183    }
184    match kind {
185        Some(1) => {}
186        Some(0) => bail!("RSA identity keys are not supported (only Ed25519)"),
187        Some(2) => bail!("secp256k1 identity keys are not supported (only Ed25519)"),
188        Some(3) => bail!("ECDSA identity keys are not supported (only Ed25519)"),
189        _ => bail!("identity key has no type"),
190    }
191    data.context("identity key has no data")?
192        .try_into()
193        .ok()
194        .context("Ed25519 key must be 32 bytes")
195}
196
197/// The peer id of a public key protobuf: an identity multihash, since every Ed25519 key's
198/// protobuf is 36 bytes (≤ 42).
199pub fn peer_id(public_key_proto: &[u8]) -> Vec<u8> {
200    let mut out = vec![0x00, public_key_proto.len() as u8];
201    out.extend_from_slice(public_key_proto);
202    out
203}
204
205const B58: &[u8; 58] = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";
206
207pub fn base58(b: &[u8]) -> String {
208    let zeros = b.iter().take_while(|&&x| x == 0).count();
209    let mut digits: Vec<u8> = Vec::new();
210    for &byte in b {
211        let mut carry = u32::from(byte);
212        for d in digits.iter_mut() {
213            carry += u32::from(*d) << 8;
214            *d = (carry % 58) as u8;
215            carry /= 58;
216        }
217        while carry > 0 {
218            digits.push((carry % 58) as u8);
219            carry /= 58;
220        }
221    }
222    let mut s = String::with_capacity(zeros + digits.len());
223    s.extend(std::iter::repeat_n('1', zeros));
224    s.extend(digits.iter().rev().map(|&d| B58[d as usize] as char));
225    s
226}
227
228pub fn unbase58(s: &str) -> Result<Vec<u8>> {
229    ensure!(s.len() <= 128, "base58 string too long");
230    let zeros = s.bytes().take_while(|&c| c == b'1').count();
231    let mut bytes: Vec<u8> = Vec::new();
232    for c in s.bytes() {
233        let mut carry =
234            B58.iter()
235                .position(|&x| x == c)
236                .with_context(|| format!("{:?} is not base58", c as char))? as u32;
237        for b in bytes.iter_mut() {
238            carry += u32::from(*b) * 58;
239            *b = (carry & 0xff) as u8;
240            carry >>= 8;
241        }
242        while carry > 0 {
243            bytes.push((carry & 0xff) as u8);
244            carry >>= 8;
245        }
246    }
247    let mut out = vec![0u8; zeros];
248    out.extend(bytes.iter().rev());
249    Ok(out)
250}
251
252pub fn peer_id_string(peer_id: &[u8]) -> String {
253    base58(peer_id)
254}
255
256// ---------------------------------------------------------------------------------------
257// Multiaddrs: the /ip4|ip6/…/tcp/…[/p2p/…] subset.
258
259const P_IP4: u64 = 0x04;
260const P_TCP: u64 = 0x06;
261const P_IP6: u64 = 0x29;
262const P_P2P: u64 = 0x01a5;
263
264pub fn multiaddr_bytes(addr: &SocketAddr) -> Vec<u8> {
265    let mut out = Vec::new();
266    match addr.ip() {
267        IpAddr::V4(a) => {
268            put_uvarint(&mut out, P_IP4);
269            out.extend_from_slice(&a.octets());
270        }
271        IpAddr::V6(a) => {
272            put_uvarint(&mut out, P_IP6);
273            out.extend_from_slice(&a.octets());
274        }
275    }
276    put_uvarint(&mut out, P_TCP);
277    out.extend_from_slice(&addr.port().to_be_bytes());
278    out
279}
280
281pub fn multiaddr_string(addr: &SocketAddr) -> String {
282    match addr.ip() {
283        IpAddr::V4(a) => format!("/ip4/{a}/tcp/{}", addr.port()),
284        IpAddr::V6(a) => format!("/ip6/{a}/tcp/{}", addr.port()),
285    }
286}
287
288/// A binary multiaddr as text; components outside the subset are shown by code.
289pub fn multiaddr_text(mut b: &[u8]) -> Result<String> {
290    let mut s = String::new();
291    while !b.is_empty() {
292        let (code, n) = get_uvarint(b)?;
293        b = &b[n..];
294        let take = |b: &mut &[u8], n: usize| -> Result<Vec<u8>> {
295            ensure!(b.len() >= n, "truncated multiaddr");
296            let v = b[..n].to_vec();
297            *b = &b[n..];
298            Ok(v)
299        };
300        match code {
301            P_IP4 => {
302                let v = take(&mut b, 4)?;
303                s.push_str(&format!("/ip4/{}.{}.{}.{}", v[0], v[1], v[2], v[3]));
304            }
305            P_IP6 => {
306                let v: [u8; 16] = take(&mut b, 16)?.try_into().unwrap_or([0; 16]);
307                s.push_str(&format!("/ip6/{}", std::net::Ipv6Addr::from(v)));
308            }
309            P_TCP | 0x0111 => {
310                let v = take(&mut b, 2)?;
311                let name = if code == P_TCP { "tcp" } else { "udp" };
312                s.push_str(&format!("/{name}/{}", u16::from_be_bytes([v[0], v[1]])));
313            }
314            0x01cc => s.push_str("/quic-v1"),
315            0x01cd => s.push_str("/quic"),
316            P_P2P => {
317                let (len, n) = get_uvarint(b)?;
318                b = &b[n..];
319                let v = take(&mut b, len as usize)?;
320                s.push_str(&format!("/p2p/{}", base58(&v)));
321            }
322            other => {
323                // Unknown component of unknown length: stop rather than misread the rest.
324                s.push_str(&format!("/<{other:#x}>"));
325                break;
326            }
327        }
328    }
329    Ok(s)
330}
331
332/// A text multiaddr in binary: `/ip4|ip6/…`, `/tcp|udp/…`, `/quic-v1`, `/p2p/<id>`, in any
333/// order the text gives them. Anything else is refused rather than guessed.
334pub fn multiaddr_from_text(s: &str) -> Result<Vec<u8>> {
335    ensure!(
336        s.starts_with('/') && s.len() <= 256,
337        "{s:?} is not a multiaddr"
338    );
339    let mut out = Vec::new();
340    let mut parts = s.trim_end_matches('/').split('/').skip(1);
341    while let Some(name) = parts.next() {
342        let mut value = || {
343            parts
344                .next()
345                .with_context(|| format!("{name} needs a value in {s}"))
346        };
347        match name {
348            "ip4" => {
349                let a: std::net::Ipv4Addr = value()?.parse().context("bad ip4 address")?;
350                put_uvarint(&mut out, P_IP4);
351                out.extend_from_slice(&a.octets());
352            }
353            "ip6" => {
354                let a: std::net::Ipv6Addr = value()?.parse().context("bad ip6 address")?;
355                put_uvarint(&mut out, P_IP6);
356                out.extend_from_slice(&a.octets());
357            }
358            "tcp" | "udp" => {
359                let port: u16 = value()?.parse().context("bad port")?;
360                put_uvarint(&mut out, if name == "tcp" { P_TCP } else { 0x0111 });
361                out.extend_from_slice(&port.to_be_bytes());
362            }
363            "quic-v1" => put_uvarint(&mut out, 0x01cc),
364            "p2p" | "ipfs" => {
365                let id = unbase58(value()?).context("the /p2p/ peer id is not base58")?;
366                put_uvarint(&mut out, P_P2P);
367                put_uvarint(&mut out, id.len() as u64);
368                out.extend(id);
369            }
370            other => bail!("multiaddr component /{other} is not supported in {s}"),
371        }
372    }
373    ensure!(!out.is_empty(), "empty multiaddr");
374    Ok(out)
375}
376
377/// `host:port`, `/ip4/a/tcp/p[/p2p/id]` or `/ip6/…`: the address to dial and the expected
378/// peer id, if named.
379pub fn parse_dial_target(s: &str) -> Result<(String, Option<Vec<u8>>)> {
380    if !s.starts_with('/') {
381        return Ok((s.to_string(), None));
382    }
383    let parts: Vec<&str> = s.trim_end_matches('/').split('/').skip(1).collect();
384    let (host, rest) = match parts.as_slice() {
385        ["ip4", ip, "tcp", port, rest @ ..] => (format!("{ip}:{port}"), rest),
386        ["ip6", ip, "tcp", port, rest @ ..] => (format!("[{ip}]:{port}"), rest),
387        ["dns4" | "dns6" | "dns", name, "tcp", port, rest @ ..] => (format!("{name}:{port}"), rest),
388        _ => bail!("only /ip4|ip6|dns/…/tcp/… multiaddrs can be dialled: {s}"),
389    };
390    let peer = match rest {
391        [] => None,
392        ["p2p" | "ipfs", id] => Some(unbase58(id).context("the /p2p/ peer id is not base58")?),
393        _ => bail!("unsupported multiaddr suffix in {s}"),
394    };
395    Ok((host, peer))
396}
397
398// ---------------------------------------------------------------------------------------
399// multistream-select 1.0
400
401async fn ms_read(io: &mut impl Io) -> Result<String> {
402    let b = read_length_prefixed(io, MAX_MULTISTREAM_MESSAGE).await?;
403    let text = String::from_utf8(b).context("multistream message is not UTF-8")?;
404    text.strip_suffix('\n')
405        .map(str::to_string)
406        .context("multistream message does not end in a newline")
407}
408
409fn ms_message(s: &str) -> Vec<u8> {
410    length_prefixed(format!("{s}\n").as_bytes())
411}
412
413/// Answer a dialer's proposals until one is supported; returns it.
414pub async fn ms_listen(io: &mut impl Io, supported: impl Fn(&str) -> bool) -> Result<String> {
415    let header = ms_read(io).await?;
416    ensure!(
417        header == MULTISTREAM,
418        "expected {MULTISTREAM}, got {header:?}"
419    );
420    io.write_all(&ms_message(MULTISTREAM)).await?;
421    for _ in 0..MAX_PROPOSALS {
422        let proposal = ms_read(io).await?;
423        if supported(&proposal) {
424            io.write_all(&ms_message(&proposal)).await?;
425            return Ok(proposal);
426        }
427        io.write_all(&ms_message("na")).await?;
428    }
429    bail!("no supported protocol in {MAX_PROPOSALS} proposals")
430}
431
432/// Propose `protocol`; Ok(true) when the listener accepts it, Ok(false) on "na".
433pub async fn ms_dial(io: &mut impl Io, protocol: &str) -> Result<bool> {
434    let mut out = ms_message(MULTISTREAM);
435    out.extend(ms_message(protocol));
436    io.write_all(&out).await?;
437    let header = ms_read(io).await?;
438    ensure!(
439        header == MULTISTREAM,
440        "expected {MULTISTREAM}, got {header:?}"
441    );
442    let answer = ms_read(io).await?;
443    if answer == protocol {
444        Ok(true)
445    } else if answer == "na" {
446        Ok(false)
447    } else {
448        bail!("listener answered {answer:?} to {protocol:?}")
449    }
450}
451
452/// Propose each protocol in turn on one stream; the first the listener accepts, or None when
453/// it refuses them all.
454pub async fn ms_dial_any(io: &mut impl Io, protocols: &[&str]) -> Result<Option<String>> {
455    let Some((first, rest)) = protocols.split_first() else {
456        return Ok(None);
457    };
458    let mut out = ms_message(MULTISTREAM);
459    out.extend(ms_message(first));
460    io.write_all(&out).await?;
461    let header = ms_read(io).await?;
462    ensure!(
463        header == MULTISTREAM,
464        "expected {MULTISTREAM}, got {header:?}"
465    );
466    let mut proposed = *first;
467    let mut rest = rest.iter();
468    loop {
469        let answer = ms_read(io).await?;
470        if answer == proposed {
471            return Ok(Some(answer));
472        }
473        ensure!(
474            answer == "na",
475            "listener answered {answer:?} to {proposed:?}"
476        );
477        let Some(next) = rest.next() else {
478            return Ok(None);
479        };
480        proposed = next;
481        io.write_all(&ms_message(next)).await?;
482    }
483}
484
485// ---------------------------------------------------------------------------------------
486// identify
487
488pub struct Identify {
489    pub protocol_version: String,
490    pub agent_version: String,
491    pub public_key: Vec<u8>,
492    pub listen_addrs: Vec<Vec<u8>>,
493    pub observed_addr: Option<Vec<u8>>,
494    pub protocols: Vec<String>,
495}
496
497pub fn encode_identify(id: &Identify) -> Vec<u8> {
498    let mut out = Vec::new();
499    pb_bytes(&mut out, 1, &id.public_key);
500    for a in &id.listen_addrs {
501        pb_bytes(&mut out, 2, a);
502    }
503    for p in &id.protocols {
504        pb_bytes(&mut out, 3, p.as_bytes());
505    }
506    if let Some(o) = &id.observed_addr {
507        pb_bytes(&mut out, 4, o);
508    }
509    pb_bytes(&mut out, 5, id.protocol_version.as_bytes());
510    pb_bytes(&mut out, 6, id.agent_version.as_bytes());
511    out
512}
513
514pub fn decode_identify(b: &[u8]) -> Result<Identify> {
515    let mut id = Identify {
516        protocol_version: String::new(),
517        agent_version: String::new(),
518        public_key: Vec::new(),
519        listen_addrs: Vec::new(),
520        observed_addr: None,
521        protocols: Vec::new(),
522    };
523    let text = |v: &[u8]| String::from_utf8_lossy(v).into_owned();
524    for (f, v) in pb_fields(b)? {
525        if let PbValue::Bytes(v) = v {
526            match f {
527                1 => id.public_key = v.to_vec(),
528                2 => id.listen_addrs.push(v.to_vec()),
529                3 => id.protocols.push(text(v)),
530                4 => id.observed_addr = Some(v.to_vec()),
531                5 => id.protocol_version = text(v),
532                6 => id.agent_version = text(v),
533                _ => {}
534            }
535        }
536    }
537    Ok(id)
538}
539
540/// Message bytes as the model sees them: text, or hex when they are not printable UTF-8.
541pub fn shown(b: &[u8]) -> (String, &'static str) {
542    match std::str::from_utf8(b) {
543        Ok(s)
544            if !s
545                .chars()
546                .any(|c| c.is_control() && c != '\n' && c != '\r' && c != '\t') =>
547        {
548            (s.to_string(), "utf8")
549        }
550        _ => (hex::encode(b), "hex"),
551    }
552}
553
554/// An action's `data` in its declared `encoding` (utf8 by default, or hex).
555pub fn data_bytes(v: &serde_json::Value) -> Result<Vec<u8>> {
556    let data = v["data"].as_str().context("data is required")?;
557    match v["encoding"].as_str().unwrap_or("utf8") {
558        "utf8" => Ok(data.as_bytes().to_vec()),
559        "hex" => hex::decode(data).context("data is not valid hex"),
560        e => bail!("encoding must be utf8 or hex, not {e:?}"),
561    }
562}
netget/server/libp2p/wire.rs

netget/server/libp2p/
wire.rs
