netget/server/adc/
codec.rs

1//! ADC BASE/TIGR hub negotiation and selected chat/search routing.
2use 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 std::{
7    collections::HashMap,
8    sync::{
9        atomic::{AtomicU32, Ordering},
10        Arc, Mutex,
11    },
12};
13use tokio::{io::AsyncWriteExt, sync::mpsc};
14pub const MAX_COMMAND: usize = 64 * 1024;
15pub fn escape(text: &str) -> Result<String> {
16    ensure!(
17        text.len() <= 8192 && !text.contains('\0'),
18        "ADC field too large or contains NUL"
19    );
20    Ok(text
21        .replace('\\', "\\\\")
22        .replace(' ', "\\s")
23        .replace('\n', "\\n"))
24}
25pub fn unescape(text: &str) -> Result<String> {
26    let mut out = String::new();
27    let mut chars = text.chars();
28    while let Some(ch) = chars.next() {
29        if ch == '\\' {
30            out.push(match chars.next() {
31                Some('\\') => '\\',
32                Some('s') => ' ',
33                Some('n') => '\n',
34                _ => bail!("invalid ADC escape"),
35            });
36        } else {
37            out.push(ch);
38        }
39    }
40    ensure!(!out.contains(['\0', '\r']), "invalid ADC field");
41    Ok(out)
42}
43pub fn words(frame: &[u8]) -> Result<Vec<String>> {
44    ensure!(
45        frame.ends_with(b"\n") && frame.len() <= MAX_COMMAND,
46        "ADC framing"
47    );
48    let line = std::str::from_utf8(&frame[..frame.len() - 1])?;
49    ensure!(!line.contains(['\r', '\0']) && line.len() >= 4, "ADC line");
50    let words: Vec<_> = line.split(' ').map(unescape).collect::<Result<_>>()?;
51    ensure!(
52        words.len() <= 128
53            && words[0].len() == 4
54            && words[0].bytes().all(|b| b.is_ascii_uppercase()),
55        "ADC command"
56    );
57    Ok(words)
58}
59pub fn encode(cmd: &str, args: &[String]) -> Result<Vec<u8>> {
60    let mut line = cmd.to_string();
61    for arg in args {
62        line.push(' ');
63        line.push_str(&escape(arg)?);
64    }
65    line.push('\n');
66    ensure!(line.len() <= MAX_COMMAND, "ADC command cap");
67    Ok(line.into_bytes())
68}
69pub fn hash(bytes: &[u8]) -> String {
70    use tiger::{Digest, Tiger};
71    crate::server::p2p_support::base32(&Tiger::digest(bytes))
72}
73fn sid(value: u32) -> String {
74    let v = (value & 0xfffff) << 4;
75    crate::server::p2p_support::base32(&v.to_be_bytes()[1..])[..4].to_string()
76}
77pub fn valid_sid(text: &str) -> bool {
78    text.len() == 4
79        && text
80            .bytes()
81            .all(|b| b.is_ascii_uppercase() || (b'2'..=b'7').contains(&b))
82}
83pub fn cid(text: &str) -> Result<Vec<u8>> {
84    ensure!(text.len() == 39, "ADC CID/PID length");
85    let mut out = Vec::new();
86    let mut value = 0u32;
87    let mut bits = 0;
88    for ch in text.bytes() {
89        let n = match ch {
90            b'A'..=b'Z' => ch - b'A',
91            b'2'..=b'7' => ch - b'2' + 26,
92            _ => bail!("invalid ADC base32"),
93        };
94        value = (value << 5) | u32::from(n);
95        bits += 5;
96        if bits >= 8 {
97            bits -= 8;
98            out.push((value >> bits) as u8);
99        }
100    }
101    ensure!(
102        out.len() == 24 && (value & 7) == 0,
103        "noncanonical ADC identity"
104    );
105    Ok(out)
106}
107struct User {
108    nickname: String,
109    info: Vec<u8>,
110    tx: mpsc::Sender<Vec<u8>>,
111    stop: tokio_util::sync::CancellationToken,
112}
113#[derive(Default)]
114pub struct Hub {
115    next: AtomicU32,
116    users: Mutex<HashMap<String, User>>,
117}
118impl Hub {
119    fn broadcast(&self, except: &str, bytes: Vec<u8>) {
120        if let Ok(users) = self.users.lock() {
121            for (id, user) in users.iter() {
122                if id != except {
123                    if user.tx.try_send(bytes.clone()).is_err() {
124                        user.stop.cancel();
125                    }
126                }
127            }
128        }
129    }
130}
131pub struct Device {
132    frame: Framer,
133    hub: Arc<Hub>,
134    sid: String,
135    supported: bool,
136    joined: bool,
137    tx: mpsc::Sender<Vec<u8>>,
138    rx: Option<mpsc::Receiver<Vec<u8>>>,
139    pending: Option<(String, Vec<u8>)>,
140    stop: tokio_util::sync::CancellationToken,
141}
142impl Device {
143    pub fn new(hub: Arc<Hub>) -> Self {
144        let (tx, rx) = mpsc::channel(32);
145        let id = sid(hub.next.fetch_add(1, Ordering::Relaxed));
146        Self {
147            frame: Framer::default(),
148            hub,
149            sid: id,
150            supported: false,
151            joined: false,
152            tx,
153            rx: Some(rx),
154            pending: None,
155            stop: tokio_util::sync::CancellationToken::new(),
156        }
157    }
158}
159impl Default for Device {
160    fn default() -> Self {
161        Self::new(Arc::new(Hub::default()))
162    }
163}
164impl Drop for Device {
165    fn drop(&mut self) {
166        if self.joined {
167            if let Ok(mut users) = self.hub.users.lock() {
168                users.remove(&self.sid);
169            }
170            self.hub
171                .broadcast(&self.sid, format!("IQUI {}\n", self.sid).into_bytes());
172        }
173    }
174}
175#[async_trait]
176impl DeviceSession for Device {
177    fn outgoing(&mut self) -> Option<mpsc::Receiver<Vec<u8>>> {
178        self.rx.take()
179    }
180    fn greeting(&mut self) -> Result<Vec<u8>> {
181        Ok(format!(
182            "ISUP ADBASE ADTIGR\nISID {}\nIINF NINetGet DESelected\\shub\\ssimulator VE1.0\n",
183            self.sid
184        )
185        .into_bytes())
186    }
187    async fn read(&mut self, s: &mut ReadStream) -> Result<Vec<u8>> {
188        tokio::select! {r=self.frame.delimited(s,b'\n',MAX_COMMAND)=>r,_=self.stop.cancelled()=>bail!("slow hub peer disconnected")}
189    }
190    fn receive(&mut self, b: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
191        let w = words(b)?;
192        if w[0] == "HSUP" {
193            ensure!(
194                !self.supported
195                    && w.iter().any(|v| v == "ADBASE")
196                    && w.iter().any(|v| v == "ADTIGR"),
197                "ADC BASE/TIGR required"
198            );
199            self.supported = true;
200            return Ok((vec![], None));
201        }
202        ensure!(
203            self.supported && w.get(1) == Some(&self.sid),
204            "ADC negotiation or SID mismatch"
205        );
206        if w[0] == "BINF" && !self.joined {
207            let tags = |tag: &str| w.iter().skip(2).find_map(|word| word.strip_prefix(tag));
208            let pid = cid(tags("PD").context("ADC PID required")?)?;
209            let public = tags("ID").context("ADC CID required")?;
210            ensure!(hash(&pid) == public, "CID/PID mismatch");
211            let nick = tags("NI").context("nickname required")?;
212            ensure!(
213                !nick.is_empty() && nick.len() <= 64 && !nick.contains(['\n', '\r']),
214                "invalid nickname"
215            );
216            let args: Vec<_> = w
217                .iter()
218                .skip(1)
219                .filter(|v| !v.starts_with("PD"))
220                .cloned()
221                .collect();
222            self.pending = Some((nick.into(), encode("BINF", &args)?));
223            return Ok((
224                vec![],
225                Some(json!({"operation":"identify","sid":self.sid,"cid":public,"nickname":nick})),
226            ));
227        }
228        ensure!(self.joined, "ADC identification not accepted");
229        let operation = match w[0].as_str() {
230            "BMSG" | "DMSG" => "chat",
231            "BSCH" => "search",
232            "DRES" => "search_result",
233            "DCTM" => "connect",
234            "DRCM" => "reverse_connect",
235            _ => return Ok((b"ISTA 140 Unsupported\n".to_vec(), None)),
236        };
237        let direct = w[0].starts_with('D');
238        if direct {
239            ensure!(
240                w.len() >= 4 && valid_sid(&w[2]),
241                "invalid direct destination"
242            );
243        }
244        let args = &w[if direct { 3 } else { 2 }..];
245        match operation {
246            "connect" => {
247                ensure!(
248                    args.len() == 3 && matches!(args[0].as_str(), "ADC/1.0" | "ADCS/1.0"),
249                    "connect syntax"
250                );
251                let port: u16 = args[1].parse()?;
252                ensure!(
253                    port > 0 && !args[2].is_empty() && args[2].len() <= 128,
254                    "connect endpoint/token"
255                );
256            }
257            "reverse_connect" => ensure!(
258                args.len() == 2
259                    && matches!(args[0].as_str(), "ADC/1.0" | "ADCS/1.0")
260                    && !args[1].is_empty()
261                    && args[1].len() <= 128,
262                "reverse-connect syntax"
263            ),
264            "chat" => ensure!(!args.is_empty(), "chat message required"),
265            "search" | "search_result" => ensure!(
266                !args.is_empty() && args.iter().all(|a| a.len() >= 2),
267                "search tags required"
268            ),
269            _ => {}
270        }
271        Ok((
272            vec![],
273            Some(
274                json!({"operation":operation,"sid":self.sid,"destination":if direct{Some(&w[2])}else{None},"arguments":w.iter().skip(if direct{3}else{2}).collect::<Vec<_>>(),"wire":std::str::from_utf8(b)?}),
275            ),
276        ))
277    }
278    fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
279        let accepted = a.is_some_and(|a| a["accepted"] == true);
280        if r["operation"] == "identify" {
281            if !accepted {
282                self.pending = None;
283                return Ok(b"ISTA 223 Identification\\srefused\n".to_vec());
284            }
285            let (nickname, info) = self.pending.take().context("missing pending identity")?;
286            let mut users = self
287                .hub
288                .users
289                .lock()
290                .map_err(|_| anyhow::anyhow!("hub lock"))?;
291            if users.values().any(|u| u.nickname == nickname) {
292                return Ok(b"ISTA 222 Nickname\\sin\\suse\n".to_vec());
293            }
294            ensure!(users.len() < 256, "hub full");
295            let mut initial = Vec::new();
296            for user in users.values() {
297                initial.extend_from_slice(&user.info);
298                if user.tx.try_send(info.clone()).is_err() {
299                    user.stop.cancel();
300                }
301            }
302            initial.extend_from_slice(&info);
303            users.insert(
304                self.sid.clone(),
305                User {
306                    nickname,
307                    info,
308                    tx: self.tx.clone(),
309                    stop: self.stop.clone(),
310                },
311            );
312            self.joined = true;
313            return Ok(initial);
314        }
315        if !accepted {
316            return Ok(b"ISTA 140 Operation\\srefused\n".to_vec());
317        }
318        let bytes = r["wire"].as_str().context("wire")?.as_bytes().to_vec();
319        if let Some(destination) = r["destination"].as_str() {
320            let users = self
321                .hub
322                .users
323                .lock()
324                .map_err(|_| anyhow::anyhow!("hub lock"))?;
325            if let Some(user) = users.get(destination) {
326                user.tx
327                    .try_send(bytes)
328                    .map_err(|_| anyhow::anyhow!("destination queue full"))?;
329            } else {
330                return Ok(b"ISTA 140 Unknown\\sdestination\n".to_vec());
331            }
332        } else {
333            self.hub.broadcast("", bytes);
334        }
335        Ok(vec![])
336    }
337}
338pub struct Scanner {
339    frame: Framer,
340    sid: String,
341    pid: String,
342    cid: String,
343    nick: String,
344}
345impl Default for Scanner {
346    fn default() -> Self {
347        use rand::RngCore;
348        let mut bytes = [0u8; 24];
349        rand::thread_rng().fill_bytes(&mut bytes);
350        let public = hash(&bytes);
351        Self {
352            frame: Framer::default(),
353            sid: String::new(),
354            pid: crate::server::p2p_support::base32(&bytes),
355            nick: format!("NetGet{}", &public[..6]),
356            cid: public,
357        }
358    }
359}
360pub fn validate(v: &Value) -> Result<()> {
361    match v["type"].as_str() {
362        Some("adc_chat") => {
363            escape(v["message"].as_str().context("message")?)?;
364            if let Some(sid) = v["destination"].as_str() {
365                ensure!(valid_sid(sid), "destination SID");
366            }
367        }
368        Some("adc_search") => {
369            escape(v["query"].as_str().context("query")?)?;
370        }
371        Some("adc_connect") | Some("adc_reverse_connect") => {
372            ensure!(
373                valid_sid(v["destination"].as_str().context("destination")?),
374                "destination SID"
375            );
376            if v["type"] == "adc_connect" {
377                ensure!(
378                    (1..=65535).contains(&v["port"].as_u64().context("port")?),
379                    "port"
380                );
381            }
382            escape(v["token"].as_str().context("token")?)?;
383        }
384        Some("adc_search_result") => {
385            ensure!(
386                valid_sid(v["destination"].as_str().context("destination")?),
387                "destination SID"
388            );
389            escape(v["identifier"].as_str().context("identifier")?)?;
390            crate::server::p2p_support::number(v, "size", u64::MAX)?;
391            escape(v["tth"].as_str().context("tth")?)?;
392            escape(v["token"].as_str().context("token")?)?;
393        }
394        _ => bail!("unknown ADC action"),
395    }
396    Ok(())
397}
398#[async_trait]
399impl ScannerSession for Scanner {
400    async fn open(&mut self, s: &mut Stream) -> Result<()> {
401        s.write_all(b"HSUP ADBASE ADTIGR\n").await?;
402        let mut supports = false;
403        let mut identified = false;
404        for _ in 0..64 {
405            let w = words(&self.frame.delimited(s, b'\n', MAX_COMMAND).await?)?;
406            match w[0].as_str() {
407                "ISUP" => {
408                    ensure!(
409                        w.iter().any(|v| v == "ADBASE") && w.iter().any(|v| v == "ADTIGR"),
410                        "hub lacks BASE/TIGR"
411                    );
412                    supports = true;
413                }
414                "ISID" => {
415                    ensure!(supports && w.len() == 2 && valid_sid(&w[1]), "invalid ISID");
416                    self.sid = w[1].clone();
417                    let info = encode(
418                        "BINF",
419                        &[
420                            self.sid.clone(),
421                            format!("ID{}", self.cid),
422                            format!("PD{}", self.pid),
423                            format!("NI{}", self.nick),
424                            "SS0".into(),
425                            "SF0".into(),
426                            "SL1".into(),
427                            "VENE1.0".into(),
428                            "HN1".into(),
429                            "HR0".into(),
430                            "HO0".into(),
431                        ],
432                    )?;
433                    s.write_all(&info).await?;
434                }
435                "BINF" if w.get(1) == Some(&self.sid) => identified = true,
436                "ISTA" if w.get(1).is_some_and(|code| code.starts_with('2')) => {
437                    bail!("ADC identification refused")
438                }
439                "IGPA" => bail!("password-authenticated ADC hubs are outside this selected scope"),
440                _ => {}
441            }
442            if identified {
443                return Ok(());
444            }
445        }
446        bail!("ADC handshake exceeded 64 frames")
447    }
448    fn connected(&self) -> Value {
449        json!({"sid":self.sid,"cid":self.cid,"nickname":self.nick})
450    }
451    async fn exchange(&mut self, s: &mut Stream, a: &Value) -> Result<Value> {
452        validate(a)?;
453        let bytes = match a["type"].as_str().context("type")? {
454            "adc_chat" => {
455                if let Some(dest) = a["destination"].as_str() {
456                    encode(
457                        "DMSG",
458                        &[
459                            self.sid.clone(),
460                            dest.into(),
461                            a["message"].as_str().context("message")?.into(),
462                        ],
463                    )?
464                } else {
465                    encode(
466                        "BMSG",
467                        &[
468                            self.sid.clone(),
469                            a["message"].as_str().context("message")?.into(),
470                        ],
471                    )?
472                }
473            }
474            "adc_search" => encode(
475                "BSCH",
476                &[
477                    self.sid.clone(),
478                    format!("AN{}", a["query"].as_str().context("query")?),
479                    "TO1".into(),
480                ],
481            )?,
482            "adc_connect" => encode(
483                "DCTM",
484                &[
485                    self.sid.clone(),
486                    a["destination"].as_str().context("destination")?.into(),
487                    if a["secure"] == true {
488                        "ADCS/1.0".into()
489                    } else {
490                        "ADC/1.0".into()
491                    },
492                    a["port"].to_string(),
493                    a["token"].as_str().context("token")?.into(),
494                ],
495            )?,
496            "adc_reverse_connect" => encode(
497                "DRCM",
498                &[
499                    self.sid.clone(),
500                    a["destination"].as_str().context("destination")?.into(),
501                    if a["secure"] == true {
502                        "ADCS/1.0".into()
503                    } else {
504                        "ADC/1.0".into()
505                    },
506                    a["token"].as_str().context("token")?.into(),
507                ],
508            )?,
509            "adc_search_result" => encode(
510                "DRES",
511                &[
512                    self.sid.clone(),
513                    a["destination"].as_str().context("destination")?.into(),
514                    format!("FN{}", a["identifier"].as_str().context("identifier")?),
515                    format!("SI{}", a["size"]),
516                    format!("TR{}", a["tth"].as_str().context("tth")?),
517                    format!("TO{}", a["token"].as_str().context("token")?),
518                ],
519            )?,
520            _ => bail!("ADC action"),
521        };
522        s.write_all(&bytes).await?;
523        Ok(json!({"sent":true}))
524    }
525    async fn idle(&mut self, s: &mut Stream) -> Result<Option<Value>> {
526        let w = words(&self.frame.delimited(s, b'\n', MAX_COMMAND).await?)?;
527        Ok(Some(json!({"command":w[0],"arguments":&w[1..]})))
528    }
529}
netget/server/adc/codec.rs

netget/server/adc/
codec.rs
