1use crate::server::p2p_support::{DeviceSession, Framer, ReadStream, ScannerSession, Stream};
3use anyhow::{bail, ensure, Context, Result};
4use async_trait::async_trait;
5use serde_json::{json, Value};
6use tokio::io::{AsyncRead, AsyncWriteExt};
7pub const MAX_COMMAND: usize = 65536 + 23;
8const HEADERS: &str =
9 "User-Agent: NetGet/0.3\r\nX-Ultrapeer: False\r\nAccept: application/x-gnutella-packets\r\n";
10fn size(b: &[u8]) -> Result<usize> {
11 let n = u32::from_le_bytes(b[19..23].try_into()?) as usize;
12 ensure!(
13 n <= 65536 && b[17] <= 16 && b[18] <= 16 && b[17] as u16 + b[18] as u16 <= 16,
14 "invalid descriptor limits"
15 );
16 Ok(n + 23)
17}
18fn hex(b: &[u8]) -> String {
19 b.iter().map(|c| format!("{c:02x}")).collect()
20}
21fn guid(s: &str) -> Result<[u8; 16]> {
22 ensure!(
23 s.len() == 32 && s.bytes().all(|b| b.is_ascii_hexdigit()),
24 "GUID requires 32 hex digits"
25 );
26 let mut out = [0; 16];
27 for (i, v) in out.iter_mut().enumerate() {
28 *v = u8::from_str_radix(&s[i * 2..i * 2 + 2], 16)?;
29 }
30 Ok(out)
31}
32fn text(s: &str) -> Result<()> {
33 ensure!(
34 s.len() <= 4096 && !s.contains(['\0', '\r', '\n']),
35 "descriptor string"
36 );
37 Ok(())
38}
39fn endpoint(v: &Value) -> Result<Vec<u8>> {
40 let port = v["port"].as_u64().unwrap_or(6346);
41 ensure!((1..=65535).contains(&port), "port");
42 let ip: std::net::Ipv4Addr = v["ip"].as_str().unwrap_or("127.0.0.1").parse()?;
43 let mut b = (port as u16).to_le_bytes().to_vec();
44 b.extend(ip.octets());
45 Ok(b)
46}
47fn number(v: &Value, k: &str) -> Result<u32> {
48 let n = v[k].as_u64().unwrap_or(0);
49 ensure!(n <= u32::MAX as u64, "{k}");
50 Ok(n as u32)
51}
52fn packet(id: [u8; 16], kind: u8, ttl: u8, hops: u8, data: Vec<u8>) -> Result<Vec<u8>> {
53 ensure!(
54 data.len() <= 65536 && ttl <= 16 && hops <= 16 && ttl as u16 + hops as u16 <= 16,
55 "descriptor cap"
56 );
57 let mut b = id.to_vec();
58 b.extend([kind, ttl, hops]);
59 b.extend((data.len() as u32).to_le_bytes());
60 b.extend(data);
61 Ok(b)
62}
63fn command(a: &Value) -> Result<(u8, Vec<u8>)> {
64 match a["type"].as_str().context("type")? {
65 "gnutella_ping" => Ok((0, vec![])),
66 "gnutella_query" => {
67 let s = a["query"].as_str().context("query")?;
68 ensure!(!s.is_empty(), "empty query");
69 text(s)?;
70 let mut b = vec![0, 0];
71 b.extend(s.as_bytes());
72 b.push(0);
73 Ok((0x80, b))
74 }
75 "gnutella_push" => {
76 let mut b = guid(a["servent_id"].as_str().context("servent_id")?)?.to_vec();
77 b.extend(number(a, "index")?.to_le_bytes());
78 let endpoint = endpoint(a)?;
79 b.extend(&endpoint[2..]);
80 b.extend(&endpoint[..2]);
81 Ok((0x40, b))
82 }
83 _ => bail!("unknown Gnutella action"),
84 }
85}
86pub fn validate(v: &Value) -> Result<()> {
87 command(v)?;
88 Ok(())
89}
90fn cstring<'a>(b: &'a [u8], pos: &mut usize) -> Result<&'a str> {
91 let suffix = b.get(*pos..).context("truncated string")?;
92 let n = suffix
93 .iter()
94 .position(|b| *b == 0)
95 .context("unterminated string")?;
96 let s = std::str::from_utf8(&suffix[..n])?;
97 text(s)?;
98 *pos += n + 1;
99 Ok(s)
100}
101fn decode(b: &[u8]) -> Result<Value> {
102 ensure!(
103 b.len() >= 23 && size(&b[..23])? == b.len(),
104 "descriptor length mismatch"
105 );
106 let p = &b[23..];
107 let mut v = json!({"guid":hex(&b[..16]),"kind":b[16],"ttl":b[17],"hops":b[18]});
108 match b[16] {
109 0 => {
110 ensure!(
111 p.is_empty() || p.first() == Some(&0xc3),
112 "invalid ping extension"
113 );
114 v["operation"] = json!("ping");
115 }
116 1 => {
117 ensure!(
118 p.len() >= 14 && (p.len() == 14 || p.get(14) == Some(&0xc3)),
119 "invalid pong size/extension"
120 );
121 v["operation"] = json!("pong");
122 v["port"] = json!(u16::from_le_bytes(p[..2].try_into()?));
123 v["ip"] = json!(std::net::Ipv4Addr::new(p[2], p[3], p[4], p[5]).to_string());
124 v["files"] = json!(u32::from_le_bytes(p[6..10].try_into()?));
125 v["kilobytes"] = json!(u32::from_le_bytes(p[10..14].try_into()?));
126 }
127 0x80 => {
128 ensure!(p.len() >= 3, "query size");
129 let mut pos = 2;
130 v["operation"] = json!("query");
131 v["query"] = json!(cstring(p, &mut pos)?);
132 if pos < p.len() {
133 v["opaque_extension_bytes"] = json!(p.len() - pos);
134 }
135 }
136 0x81 => {
137 ensure!(p.len() >= 27 && p[0] <= 32, "query-hit size/count");
138 v["operation"] = json!("query_hit");
139 v["port"] = json!(u16::from_le_bytes(p[1..3].try_into()?));
140 v["ip"] = json!(std::net::Ipv4Addr::new(p[3], p[4], p[5], p[6]).to_string());
141 v["speed"] = json!(u32::from_le_bytes(p[7..11].try_into()?));
142 let mut pos = 11;
143 let entries = &p[..p.len() - 16];
144 let mut rows = vec![];
145 for _ in 0..p[0] {
146 let fixed = entries.get(pos..pos + 8).context("truncated hit")?;
147 let index = u32::from_le_bytes(fixed[..4].try_into()?);
148 let size = u32::from_le_bytes(fixed[4..].try_into()?);
149 pos += 8;
150 let name = cstring(entries, &mut pos)?;
151 let urn = cstring(entries, &mut pos)?;
152 rows.push(json!({"index":index,"size":size,"name":name,"urn":urn}));
153 }
154 if pos < entries.len() {
155 v["opaque_extension_bytes"] = json!(entries.len() - pos);
156 }
157 v["results"] = json!(rows);
158 v["servent_id"] = json!(hex(&p[p.len() - 16..]));
159 }
160 0x40 => {
161 ensure!(p.len() == 26, "push size");
162 v["operation"] = json!("push");
163 v["servent_id"] = json!(hex(&p[..16]));
164 v["index"] = json!(u32::from_le_bytes(p[16..20].try_into()?));
165 v["ip"] = json!(std::net::Ipv4Addr::new(p[20], p[21], p[22], p[23]).to_string());
166 v["port"] = json!(u16::from_le_bytes(p[24..26].try_into()?));
167 }
168 _ => v["operation"] = json!("extension"),
169 };
170 Ok(v)
171}
172async fn headers<R: AsyncRead + Unpin>(f: &mut Framer, s: &mut R, expected: &str) -> Result<()> {
173 let first = f.delimited(s, b'\n', 8192).await?;
174 ensure!(
175 first == format!("{expected}\r\n").as_bytes(),
176 "Gnutella handshake refused: {}",
177 String::from_utf8_lossy(&first).trim()
178 );
179 let mut total = first.len();
180 for _ in 0..32 {
181 let line = f.delimited(s, b'\n', 8192).await?;
182 total += line.len();
183 ensure!(total <= 8192, "header cap");
184 if line == b"\r\n" {
185 return Ok(());
186 }
187 let text = std::str::from_utf8(&line)?;
188 ensure!(
189 text.ends_with("\r\n") && text.contains(':') && !text.contains('\0'),
190 "invalid handshake header"
191 );
192 ensure!(
193 !text.to_ascii_lowercase().starts_with("content-encoding:"),
194 "compressed descriptors unsupported"
195 );
196 }
197 bail!("header count cap")
198}
199#[derive(Default)]
200pub struct Device {
201 frame: Framer,
202 phase: u8,
203}
204#[async_trait]
205impl DeviceSession for Device {
206 async fn read(&mut self, s: &mut ReadStream) -> Result<Vec<u8>> {
207 if self.phase < 2 {
208 headers(
209 &mut self.frame,
210 s,
211 if self.phase == 0 {
212 "GNUTELLA CONNECT/0.6"
213 } else {
214 "GNUTELLA/0.6 200 OK"
215 },
216 )
217 .await?;
218 Ok(vec![])
219 } else {
220 self.frame.sized(s, 23, MAX_COMMAND, size).await
221 }
222 }
223 fn receive(&mut self, b: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
224 if self.phase < 2 {
225 self.phase += 1;
226 return Ok((
227 if self.phase == 1 {
228 format!("GNUTELLA/0.6 200 OK\r\n{HEADERS}\r\n").into_bytes()
229 } else {
230 vec![]
231 },
232 None,
233 ));
234 }
235 let v = decode(b)?;
236 if v["operation"] == "extension" {
237 return Ok((vec![], None));
238 }
239 Ok((vec![], Some(v)))
240 }
241 fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
242 let Some(a) = a else { return Ok(vec![]) };
243 let id = guid(r["guid"].as_str().context("guid")?)?;
244 let (kind, data) = match r["operation"].as_str() {
245 Some("ping") => {
246 let mut b = endpoint(a)?;
247 b.extend(number(a, "files")?.to_le_bytes());
248 b.extend(number(a, "kilobytes")?.to_le_bytes());
249 (1, b)
250 }
251 Some("query") => {
252 let rows = a["results"].as_array().cloned().unwrap_or_default();
253 ensure!(rows.len() <= 32, "query hits cap");
254 let mut b = vec![rows.len() as u8];
255 b.extend(endpoint(a)?);
256 b.extend(number(a, "speed")?.to_le_bytes());
257 for row in rows {
258 b.extend(number(&row, "index")?.to_le_bytes());
259 b.extend(number(&row, "size")?.to_le_bytes());
260 for k in ["name", "urn"] {
261 let s = row[k].as_str().unwrap_or("");
262 text(s)?;
263 b.extend(s.as_bytes());
264 b.push(0);
265 }
266 }
267 b.extend([0; 16]);
268 (0x81, b)
269 }
270 _ => return Ok(vec![]),
271 };
272 packet(id, kind, 1, 0, data)
273 }
274}
275#[derive(Default)]
276pub struct Scanner {
277 frame: Framer,
278}
279#[async_trait]
280impl ScannerSession for Scanner {
281 async fn open(&mut self, s: &mut Stream) -> Result<()> {
282 s.write_all(format!("GNUTELLA CONNECT/0.6\r\n{HEADERS}\r\n").as_bytes())
283 .await?;
284 headers(&mut self.frame, s, "GNUTELLA/0.6 200 OK").await?;
285 s.write_all(format!("GNUTELLA/0.6 200 OK\r\n{HEADERS}\r\n").as_bytes())
286 .await?;
287 Ok(())
288 }
289 async fn exchange(&mut self, s: &mut Stream, a: &Value) -> Result<Value> {
290 use rand::RngCore;
291 let mut id = [0; 16];
292 rand::thread_rng().fill_bytes(&mut id);
293 let (kind, payload) = command(a)?;
294 s.write_all(&packet(id, kind, 1, 0, payload)?).await?;
295 Ok(json!({"sent":true,"guid":hex(&id)}))
296 }
297 async fn idle(&mut self, s: &mut Stream) -> Result<Option<Value>> {
298 Ok(Some(decode(
299 &self.frame.sized(s, 23, MAX_COMMAND, size).await?,
300 )?))
301 }
302}