dig_dht/
wire.rs

1//! The DHT RPC wire — the four request/response messages, `type`-tagged JSON, framed as a `u32`
2//! big-endian length prefix + JSON body (the same uniform framing dig-nat uses for its control
3//! messages).
4//!
5//! ## The four methods (RLY-style, aligned to the L7 peer-network wire)
6//!
7//! | Method | Request | Response | Purpose |
8//! |---|---|---|---|
9//! | `find_node` | `{ type:"find_node", target:<64hex key> }` | `{ type:"nodes", nodes:[Contact] }` | the `k` peers this node knows closest to `target` |
10//! | `find_providers` | `{ type:"find_providers", content_key:<64hex> }` | `{ type:"providers", providers:[ProviderRecord], closer:[Contact] }` | providers held locally + `k` closer peers if none |
11//! | `add_provider` | `{ type:"add_provider", record:ProviderRecord }` | `{ type:"add_provider_ok" }` | store the record (announce to the `k` closest) |
12//! | `ping` | `{ type:"ping", nonce:uint }` | `{ type:"pong", nonce:uint }` | liveness (echoes the nonce) |
13//!
14//! A [`Contact`] on the wire is `{ peer_id:<64hex>, addresses:[{host,port,
15//! kind}] }` — the same address shape as the L7 `dig.getPeers` peers and a [`ProviderRecord`]'s
16//! addresses, so a returned contact drops straight into a `PeerTarget` for [`dig_nat::connect`].
17//!
18//! Framing (`encode` / `decode`): a `u32` big-endian body length, then the JSON body, bounded by
19//! [`MAX_FRAMED_BODY`] to guard against a malicious length prefix. This is byte-identical to the
20//! dig-nat control framing so a node speaks one framing across the peer network.
21
22use std::io;
23
24use serde::{Deserialize, Serialize};
25use tokio::io::{AsyncRead, AsyncReadExt};
26
27use crate::record::ProviderRecord;
28use crate::routing::Contact;
29
30/// A DHT RPC **request** — one of the four methods, discriminated by `type`.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(tag = "type", rename_all = "snake_case")]
33pub enum DhtRequest {
34    /// Ask for the `k` peers the responder knows closest to `target` (an XOR-metric target key).
35    FindNode {
36        /// The 64-hex target [`Key`](crate::Key) to find close peers to.
37        target: String,
38    },
39    /// Ask for providers of `content_key`; if the responder holds none, it returns the `k` peers it
40    /// knows closest to the key so the lookup can walk closer.
41    FindProviders {
42        /// The 64-hex content [`Key`](crate::Key) to find providers for.
43        content_key: String,
44    },
45    /// Announce that the record's `provider_peer_id` holds the record's content. The responder
46    /// stores the record (it is, by construction of the announce, one of the `k` closest to the key).
47    AddProvider {
48        /// The provider record to store.
49        record: ProviderRecord,
50    },
51    /// Liveness check — the responder echoes `nonce` in its `pong`.
52    Ping {
53        /// An opaque nonce the responder echoes back, so a caller can match pong to ping.
54        nonce: u64,
55    },
56}
57
58/// A DHT RPC **response**, discriminated by `type`. Each variant is the reply to the correspondingly
59/// named [`DhtRequest`].
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
61#[serde(tag = "type", rename_all = "snake_case")]
62pub enum DhtResponse {
63    /// Reply to [`DhtRequest::FindNode`]: the `k` closest contacts the responder knows.
64    Nodes {
65        /// Up to `k` contacts, closest-first to the requested target.
66        nodes: Vec<Contact>,
67    },
68    /// Reply to [`DhtRequest::FindProviders`]: any providers the responder holds for the key, plus
69    /// (always) the `closer` contacts so a lookup can continue toward the key even when providers
70    /// are already found (more may live nearer the key).
71    Providers {
72        /// Provider records the responder holds for the content key (may be empty).
73        providers: Vec<ProviderRecord>,
74        /// The `k` closest contacts the responder knows — used to walk the lookup closer.
75        closer: Vec<Contact>,
76    },
77    /// Reply to [`DhtRequest::AddProvider`]: the record was accepted + stored.
78    AddProviderOk,
79    /// Reply to [`DhtRequest::Ping`]: the echoed nonce (proves liveness + round-trip).
80    Pong {
81        /// The nonce from the ping.
82        nonce: u64,
83    },
84    /// A responder-side error (bad request, over capacity). Advisory — a lookup treats it like an
85    /// unreachable peer and moves on.
86    Error {
87        /// A stable error code.
88        code: u32,
89        /// A human-readable message.
90        message: String,
91    },
92}
93
94/// Maximum length-prefixed body — guards against a malicious length prefix forcing a huge
95/// allocation. Provider lists at `k = 20` with a handful of addresses each are well under this.
96pub const MAX_FRAMED_BODY: usize = 256 * 1024;
97
98impl DhtRequest {
99    /// Serialize as a `u32` big-endian length prefix + JSON body.
100    pub fn encode(&self) -> Vec<u8> {
101        encode_framed(self)
102    }
103    /// Read + decode a request from `r` (the serving side).
104    pub async fn decode<R: AsyncRead + Unpin>(r: &mut R) -> io::Result<Self> {
105        decode_framed(r).await
106    }
107}
108
109impl DhtResponse {
110    /// Serialize as a `u32` big-endian length prefix + JSON body.
111    pub fn encode(&self) -> Vec<u8> {
112        encode_framed(self)
113    }
114    /// Read + decode a response from `r` (the requesting side).
115    pub async fn decode<R: AsyncRead + Unpin>(r: &mut R) -> io::Result<Self> {
116        decode_framed(r).await
117    }
118}
119
120/// Serialize `value` as a `u32` big-endian length prefix + JSON body — the uniform DHT-RPC framing.
121fn encode_framed<T: Serialize>(value: &T) -> Vec<u8> {
122    let body = serde_json::to_vec(value).expect("dht message serializes");
123    let mut out = Vec::with_capacity(4 + body.len());
124    out.extend_from_slice(&(body.len() as u32).to_be_bytes());
125    out.extend_from_slice(&body);
126    out
127}
128
129/// Read + decode a length-prefixed JSON DHT message from `r`, bounded by [`MAX_FRAMED_BODY`].
130async fn decode_framed<T: for<'de> Deserialize<'de>, R: AsyncRead + Unpin>(
131    r: &mut R,
132) -> io::Result<T> {
133    let mut len_buf = [0u8; 4];
134    r.read_exact(&mut len_buf).await?;
135    let len = u32::from_be_bytes(len_buf) as usize;
136    if len > MAX_FRAMED_BODY {
137        return Err(io::Error::new(
138            io::ErrorKind::InvalidData,
139            "dht message too large",
140        ));
141    }
142    let mut body = vec![0u8; len];
143    r.read_exact(&mut body).await?;
144    serde_json::from_slice(&body).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
145}
146
147#[cfg(test)]
148mod tests {
149    use super::*;
150    use crate::record::{AddressKind, CandidateAddr};
151    use std::io::Cursor;
152
153    #[tokio::test]
154    async fn find_node_round_trips_framed() {
155        let req = DhtRequest::FindNode {
156            target: "ab".repeat(32),
157        };
158        let bytes = req.encode();
159        let mut cur = Cursor::new(bytes);
160        let back = DhtRequest::decode(&mut cur).await.unwrap();
161        assert_eq!(req, back);
162    }
163
164    #[tokio::test]
165    async fn providers_response_round_trips() {
166        let resp = DhtResponse::Providers {
167            providers: vec![ProviderRecord {
168                content_key: "cd".repeat(32),
169                provider_peer_id: "ef".repeat(32),
170                addresses: vec![CandidateAddr::direct("203.0.113.7", 9444)],
171                expires_at: 1_719_763_200,
172            }],
173            closer: vec![Contact {
174                peer_id: "12".repeat(32),
175                addresses: vec![CandidateAddr {
176                    host: "h".into(),
177                    port: 1,
178                    kind: AddressKind::Mapped,
179                }],
180            }],
181        };
182        let mut cur = Cursor::new(resp.encode());
183        let back = DhtResponse::decode(&mut cur).await.unwrap();
184        assert_eq!(resp, back);
185    }
186
187    #[test]
188    fn request_type_tags_are_snake_case() {
189        let s = serde_json::to_string(&DhtRequest::Ping { nonce: 7 }).unwrap();
190        assert!(s.contains("\"type\":\"ping\""));
191        assert!(s.contains("\"nonce\":7"));
192        let fp = serde_json::to_string(&DhtRequest::FindProviders {
193            content_key: "00".repeat(32),
194        })
195        .unwrap();
196        assert!(fp.contains("\"type\":\"find_providers\""));
197    }
198
199    #[test]
200    fn response_type_tags_are_snake_case() {
201        assert!(serde_json::to_string(&DhtResponse::AddProviderOk)
202            .unwrap()
203            .contains("\"type\":\"add_provider_ok\""));
204        assert!(serde_json::to_string(&DhtResponse::Pong { nonce: 9 })
205            .unwrap()
206            .contains("\"type\":\"pong\""));
207    }
208
209    #[tokio::test]
210    async fn oversize_length_prefix_is_rejected() {
211        // A frame claiming a body larger than MAX_FRAMED_BODY must error, not allocate.
212        let mut buf = ((MAX_FRAMED_BODY + 1) as u32).to_be_bytes().to_vec();
213        buf.extend_from_slice(b"{}");
214        let mut cur = Cursor::new(buf);
215        let err = DhtRequest::decode(&mut cur).await.unwrap_err();
216        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
217    }
218
219    #[tokio::test]
220    async fn truncated_frame_errors() {
221        // A length prefix promising 100 bytes but only 2 present → error (not a hang / partial).
222        let mut buf = 100u32.to_be_bytes().to_vec();
223        buf.extend_from_slice(b"{}");
224        let mut cur = Cursor::new(buf);
225        assert!(DhtRequest::decode(&mut cur).await.is_err());
226    }
227}
dig_dht/wire.rs

dig_dht/
wire.rs
