netget/server/libp2p/
kad.rs

1//! The Kademlia DHT message (`/ipfs/kad/1.0.0`, and the LAN DHT's `/ipfs/lan/kad/1.0.0`):
2//! one uvarint-length-prefixed protobuf per request, answered on the same stream. Peers are
3//! carried as (peer id, binary multiaddrs); both are checked on the way in and out.
4use super::ipfs;
5use super::wire::{
6    base58, multiaddr_from_text, multiaddr_text, pb_bytes, pb_fields_bounded, pb_varint, unbase58,
7    PbValue,
8};
9use anyhow::{bail, ensure, Context, Result};
10
11pub const KAD: &str = "/ipfs/kad/1.0.0";
12pub const LAN_KAD: &str = "/ipfs/lan/kad/1.0.0";
13pub const PROTOCOLS: &[&str] = &[KAD, LAN_KAD];
14/// A DHT message is at most this long. Requests are a key and a few addresses; a reply is
15/// at most `MAX_PEERS` peers with `MAX_ADDRS` addresses each.
16pub const MAX_MESSAGE: usize = 64 * 1024;
17/// Peers one message may carry in each of its peer lists (Kademlia's k is 20).
18pub const MAX_PEERS: usize = 20;
19/// Addresses one peer entry may carry.
20pub const MAX_ADDRS: usize = 16;
21/// How long a dialler waits for a DHT answer.
22pub const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
23
24pub fn is_kad(protocol: &str) -> bool {
25    PROTOCOLS.contains(&protocol)
26}
27
28#[derive(Clone, Copy, PartialEq, Eq, Debug)]
29pub enum Kind {
30    PutValue,
31    GetValue,
32    AddProvider,
33    GetProviders,
34    FindNode,
35    Ping,
36}
37
38impl Kind {
39    pub fn code(self) -> u64 {
40        match self {
41            Kind::PutValue => 0,
42            Kind::GetValue => 1,
43            Kind::AddProvider => 2,
44            Kind::GetProviders => 3,
45            Kind::FindNode => 4,
46            Kind::Ping => 5,
47        }
48    }
49    pub fn from_code(c: u64) -> Result<Kind> {
50        Ok(match c {
51            0 => Kind::PutValue,
52            1 => Kind::GetValue,
53            2 => Kind::AddProvider,
54            3 => Kind::GetProviders,
55            4 => Kind::FindNode,
56            5 => Kind::Ping,
57            other => bail!("unknown DHT message type {other}"),
58        })
59    }
60    pub fn name(self) -> &'static str {
61        match self {
62            Kind::PutValue => "PUT_VALUE",
63            Kind::GetValue => "GET_VALUE",
64            Kind::AddProvider => "ADD_PROVIDER",
65            Kind::GetProviders => "GET_PROVIDERS",
66            Kind::FindNode => "FIND_NODE",
67            Kind::Ping => "PING",
68        }
69    }
70}
71
72/// A peer as the DHT carries it.
73#[derive(Clone, Debug, PartialEq)]
74pub struct Peer {
75    pub id: Vec<u8>,
76    pub addrs: Vec<Vec<u8>>,
77}
78
79impl Peer {
80    pub fn to_json(&self) -> serde_json::Value {
81        serde_json::json!({
82            "peer_id": base58(&self.id),
83            "addrs": self.addrs.iter().filter_map(|a| multiaddr_text(a).ok()).collect::<Vec<_>>(),
84        })
85    }
86    /// From `{peer_id, addrs}` as the model writes it.
87    pub fn from_text(id: &str, addrs: &[String]) -> Result<Peer> {
88        let id = parse_peer_id(id)?;
89        ensure!(
90            addrs.len() <= MAX_ADDRS,
91            "a peer may carry at most {MAX_ADDRS} addresses"
92        );
93        let addrs = addrs
94            .iter()
95            .map(|a| multiaddr_from_text(a))
96            .collect::<Result<Vec<_>>>()?;
97        Ok(Peer { id, addrs })
98    }
99}
100
101/// A peer id in text, checked to be a multihash (identity or sha2-256), as every libp2p peer
102/// id is.
103pub fn parse_peer_id(text: &str) -> Result<Vec<u8>> {
104    let b = unbase58(text).with_context(|| format!("{text:?} is not a base58 peer id"))?;
105    check_peer_id(&b).with_context(|| format!("{text:?} is not a peer id"))?;
106    Ok(b)
107}
108
109fn check_peer_id(b: &[u8]) -> Result<()> {
110    ensure!(b.len() >= 2, "too short");
111    match b[0] {
112        0x00 => ensure!(
113            usize::from(b[1]) == b.len() - 2 && b.len() <= 44,
114            "malformed identity multihash"
115        ),
116        0x12 => ensure!(b[1] == 32 && b.len() == 34, "malformed sha2-256 multihash"),
117        other => bail!("multihash {other:#x} is not a peer id"),
118    }
119    Ok(())
120}
121
122#[derive(Clone, Debug, PartialEq)]
123pub struct Message {
124    pub kind: Kind,
125    pub key: Vec<u8>,
126    pub closer: Vec<Peer>,
127    pub providers: Vec<Peer>,
128    /// Whether a record (PUT_VALUE/GET_VALUE) was present; records are not supported.
129    pub has_record: bool,
130}
131
132impl Message {
133    pub fn request(kind: Kind, key: Vec<u8>) -> Message {
134        Message {
135            kind,
136            key,
137            closer: Vec::new(),
138            providers: Vec::new(),
139            has_record: false,
140        }
141    }
142    /// The key as text: a peer id for FIND_NODE, the multihash (base58, the CIDv0 spelling of
143    /// a sha2-256 key) otherwise.
144    pub fn key_text(&self) -> String {
145        base58(&self.key)
146    }
147    /// For a provider key: the CIDs (v0, v1 dag-pb, v1 raw) its multihash may stand for.
148    pub fn key_cids(&self) -> Vec<String> {
149        match ipfs::parse_multihash(&self.key) {
150            Ok((digest, n)) if n == self.key.len() => ipfs::cids_for_digest(&digest),
151            _ => Vec::new(),
152        }
153    }
154}
155
156fn decode_peer(b: &[u8]) -> Result<Peer> {
157    let mut p = Peer {
158        id: Vec::new(),
159        addrs: Vec::new(),
160    };
161    for (f, v) in pb_fields_bounded(b, MAX_ADDRS + 4)? {
162        match (f, v) {
163            (1, PbValue::Bytes(id)) => p.id = id.to_vec(),
164            (2, PbValue::Bytes(a)) => {
165                ensure!(
166                    p.addrs.len() < MAX_ADDRS,
167                    "a DHT peer carries more than {MAX_ADDRS} addresses"
168                );
169                p.addrs.push(a.to_vec())
170            }
171            (3, PbValue::Varint(_)) => {}
172            _ => bail!("malformed DHT peer"),
173        }
174    }
175    check_peer_id(&p.id).context("a DHT peer's id is not a peer id")?;
176    Ok(p)
177}
178
179pub fn decode(b: &[u8]) -> Result<Message> {
180    ensure!(
181        b.len() <= MAX_MESSAGE,
182        "DHT message over {MAX_MESSAGE} bytes"
183    );
184    let mut kind = None;
185    let mut m = Message::request(Kind::Ping, Vec::new());
186    for (f, v) in pb_fields_bounded(b, 2 * MAX_PEERS + 8)? {
187        match (f, v) {
188            (1, PbValue::Varint(k)) => kind = Some(Kind::from_code(k)?),
189            (2, PbValue::Bytes(k)) => {
190                ensure!(k.len() <= 256, "DHT key longer than 256 bytes");
191                m.key = k.to_vec()
192            }
193            (3, PbValue::Bytes(_)) => m.has_record = true,
194            (8, PbValue::Bytes(p)) => {
195                ensure!(
196                    m.closer.len() < MAX_PEERS,
197                    "more than {MAX_PEERS} closer peers"
198                );
199                m.closer.push(decode_peer(p)?)
200            }
201            (9, PbValue::Bytes(p)) => {
202                ensure!(
203                    m.providers.len() < MAX_PEERS,
204                    "more than {MAX_PEERS} provider peers"
205                );
206                m.providers.push(decode_peer(p)?)
207            }
208            (10, PbValue::Varint(_)) => {}
209            _ => bail!("malformed DHT message (field {f})"),
210        }
211    }
212    m.kind = kind.unwrap_or(Kind::PutValue);
213    Ok(m)
214}
215
216fn encode_peer(p: &Peer) -> Vec<u8> {
217    let mut out = Vec::new();
218    pb_bytes(&mut out, 1, &p.id);
219    for a in &p.addrs {
220        pb_bytes(&mut out, 2, a);
221    }
222    out
223}
224
225pub fn encode(m: &Message) -> Vec<u8> {
226    let mut out = Vec::new();
227    // proto3: a zero enum is the default and is not written.
228    if m.kind.code() != 0 {
229        pb_varint(&mut out, 1, m.kind.code());
230    }
231    if !m.key.is_empty() {
232        pb_bytes(&mut out, 2, &m.key);
233    }
234    for p in &m.closer {
235        pb_bytes(&mut out, 8, &encode_peer(p));
236    }
237    for p in &m.providers {
238        pb_bytes(&mut out, 9, &encode_peer(p));
239    }
240    out
241}
netget/server/libp2p/kad.rs

netget/server/libp2p/
kad.rs
