netget/server/dc_peer/
codec.rs

1//! Selected NMDC peer negotiation and ADCGET/ADCSND binary transfers.
2pub use crate::server::p2p_support::files::{content, decoded_result, tth, MAX_FILE};
3use crate::server::p2p_support::{DeviceSession, Framer, ReadStream, ScannerSession, Stream};
4use anyhow::{bail, ensure, Context, Result};
5use async_trait::async_trait;
6use serde_json::{json, Value};
7use tokio::io::{AsyncReadExt, AsyncWriteExt};
8pub const MAX_COMMAND: usize = 64 * 1024;
9pub const LOCK: &[u8] = b"EXTENDEDPROTOCOLABCABCABCABCABCABC";
10pub const DEFAULT_LISTENER_NICKNAME: &str = "NetGet";
11pub const DEFAULT_CONNECTOR_NICKNAME: &str = "NetGetClient";
12pub fn nickname(value: &str) -> Result<&str> {
13    ensure!(
14        !value.is_empty()
15            && value.len() <= 64
16            && value
17                .bytes()
18                .all(|b| (33..127).contains(&b) && !matches!(b, b'$' | b'|' | b'<' | b'>')),
19        "invalid NMDC nickname"
20    );
21    Ok(value)
22}
23pub fn nickname_parameter(default: &str) -> crate::llm::actions::ParameterDefinition {
24    crate::llm::actions::ParameterDefinition {
25        name: "nickname".into(), type_hint: "string".into(),
26        description: "Local NMDC peer nickname; match the identity announced on the hub (1..64 printable ASCII bytes, no spaces or $|<>)".into(),
27        required: false, example: json!(default), default: Some(json!(default)),
28    }
29}
30fn field(v: &str) -> Result<&str> {
31    ensure!(
32        !v.is_empty() && v.len() <= 4096 && !v.bytes().any(|b| b <= 32 || matches!(b, b'|' | b'$')),
33        "invalid NMDC field"
34    );
35    Ok(v)
36}
37pub fn validate(v: &Value) -> Result<()> {
38    ensure!(v["type"] == "dc_peer_get", "unknown NMDC peer action");
39    field(v["identifier"].as_str().context("identifier required")?)?;
40    ensure!(v["offset"].as_u64().is_some(), "offset required");
41    let len = v["length"].as_i64().context("length required")?;
42    ensure!(
43        len == -1 || (0..=MAX_FILE as i64).contains(&len),
44        "length exceeds transfer cap"
45    );
46    Ok(())
47}
48pub fn transfer_event(command: &str) -> Result<Value> {
49    let words: Vec<_> = command.split_whitespace().collect();
50    ensure!(
51        words.len() == 5 && words[0] == "$ADCGET" && words[1] == "file",
52        "unsupported peer request"
53    );
54    field(words[2])?;
55    let offset: u64 = words[3].parse()?;
56    let length: i64 = words[4].parse()?;
57    ensure!(
58        length == -1 || (0..=MAX_FILE as i64).contains(&length),
59        "request too large"
60    );
61    Ok(json!({"operation":"get","identifier":words[2],"offset":offset,"length":length}))
62}
63#[derive(Default)]
64pub struct Device {
65    frame: Framer,
66    nick: bool,
67    nickname: Option<String>,
68    key: bool,
69    lock: bool,
70}
71impl Device {
72    pub fn with_nickname(nickname: String) -> Self {
73        Self {
74            nickname: Some(nickname),
75            ..Self::default()
76        }
77    }
78}
79#[async_trait]
80impl DeviceSession for Device {
81    fn greeting(&mut self) -> Result<Vec<u8>> {
82        Ok(format!(
83            "$MyNick {}|$Lock {} Pk=NetGet|",
84            nickname(
85                self.nickname
86                    .as_deref()
87                    .unwrap_or(DEFAULT_LISTENER_NICKNAME)
88            )?,
89            std::str::from_utf8(LOCK)?
90        )
91        .into_bytes())
92    }
93    async fn read(&mut self, s: &mut ReadStream) -> Result<Vec<u8>> {
94        self.frame.delimited(s, b'|', MAX_COMMAND).await
95    }
96    fn receive(&mut self, b: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
97        let cmd = b.strip_suffix(b"|").context("delimiter")?;
98        if let Some(nick) = cmd.strip_prefix(b"$MyNick ") {
99            ensure!(!nick.is_empty() && nick.len() <= 64, "nickname");
100            self.nick = true;
101            return Ok((vec![], None));
102        }
103        if let Some(lock) = cmd.strip_prefix(b"$Lock ") {
104            let lock = lock.split(|b| *b == b' ').next().context("lock")?;
105            let mut reply = b"$Supports ADCGet TTHF XMLBZList|$Direction Upload 1|$Key ".to_vec();
106            reply.extend(crate::client::dc::calculate_dc_key(lock)?);
107            reply.push(b'|');
108            self.lock = true;
109            return Ok((reply, None));
110        }
111        if let Some(key) = cmd.strip_prefix(b"$Key ") {
112            ensure!(
113                key == crate::client::dc::calculate_dc_key(LOCK)?,
114                "wrong key"
115            );
116            self.key = true;
117            return Ok((vec![], None));
118        }
119        if cmd.starts_with(b"$Supports ") || cmd.starts_with(b"$Direction Download ") {
120            return Ok((vec![], None));
121        }
122        ensure!(
123            self.nick && self.key && self.lock,
124            "peer negotiation incomplete"
125        );
126        let cmd = std::str::from_utf8(cmd)?;
127        match transfer_event(cmd) {
128            Ok(e) => Ok((vec![], Some(e))),
129            Err(_) => Ok((b"$Error Unsupported request|".to_vec(), None)),
130        }
131    }
132    fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
133        match content(r, a) {
134            Ok(data) => {
135                let mut reply = format!(
136                    "$ADCSND file {} {} {}|",
137                    r["identifier"].as_str().context("identifier")?,
138                    r["offset"],
139                    data.len()
140                )
141                .into_bytes();
142                reply.extend(data);
143                Ok(reply)
144            }
145            Err(_) => Ok(b"$Error File Not Available|".to_vec()),
146        }
147    }
148}
149#[derive(Default)]
150pub struct Scanner {
151    frame: Framer,
152    nickname: Option<String>,
153}
154impl Scanner {
155    pub fn with_nickname(nickname: String) -> Self {
156        Self {
157            nickname: Some(nickname),
158            ..Self::default()
159        }
160    }
161}
162#[async_trait]
163impl ScannerSession for Scanner {
164    async fn open(&mut self, s: &mut Stream) -> Result<()> {
165        s.write_all(
166            format!(
167                "$MyNick {}|$Lock {} Pk=NetGet|",
168                nickname(
169                    self.nickname
170                        .as_deref()
171                        .unwrap_or(DEFAULT_CONNECTOR_NICKNAME)
172                )?,
173                std::str::from_utf8(LOCK)?
174            )
175            .as_bytes(),
176        )
177        .await?;
178        let (mut nick, mut lock, mut key, mut direction) = (false, false, false, false);
179        for _ in 0..16 {
180            let frame = self.frame.delimited(s, b'|', MAX_COMMAND).await?;
181            let cmd = &frame[..frame.len() - 1];
182            if cmd.starts_with(b"$MyNick ") {
183                nick = true;
184            } else if let Some(challenge) = cmd.strip_prefix(b"$Lock ") {
185                let mut reply =
186                    b"$Supports ADCGet TTHF XMLBZList|$Direction Download 1|$Key ".to_vec();
187                reply.extend(crate::client::dc::calculate_dc_key(
188                    challenge.split(|b| *b == b' ').next().context("lock")?,
189                )?);
190                reply.push(b'|');
191                s.write_all(&reply).await?;
192                lock = true;
193            } else if let Some(received) = cmd.strip_prefix(b"$Key ") {
194                ensure!(
195                    received == crate::client::dc::calculate_dc_key(LOCK)?,
196                    "wrong peer key"
197                );
198                key = true;
199            } else if cmd.starts_with(b"$Direction Upload ") {
200                direction = true;
201            }
202            if nick && lock && key && direction {
203                return Ok(());
204            }
205        }
206        bail!("peer negotiation exceeded 16 messages")
207    }
208    async fn exchange(&mut self, s: &mut Stream, a: &Value) -> Result<Value> {
209        validate(a)?;
210        let id = a["identifier"].as_str().context("identifier")?;
211        s.write_all(format!("$ADCGET file {} {} {}|", id, a["offset"], a["length"]).as_bytes())
212            .await?;
213        let frame = self.frame.delimited(s, b'|', MAX_COMMAND).await?;
214        let text = std::str::from_utf8(&frame[..frame.len() - 1])?;
215        let words: Vec<_> = text.split_whitespace().collect();
216        if text.starts_with("$Error ") {
217            return Ok(json!({"error":&text[7..]}));
218        }
219        ensure!(
220            words.len() == 5 && words[0] == "$ADCSND" && words[1] == "file" && words[2] == id,
221            "unexpected transfer response"
222        );
223        ensure!(
224            words[3].parse::<u64>()? == a["offset"].as_u64().context("offset")?,
225            "offset mismatch"
226        );
227        let count: usize = words[4].parse()?;
228        ensure!(count <= MAX_FILE, "transfer exceeds cap");
229        if let Some(len) = a["length"].as_i64().filter(|n| *n >= 0) {
230            ensure!(count == len as usize, "length mismatch");
231        }
232        let mut data = vec![0; count];
233        s.read_exact(&mut data).await?;
234        decoded_result(id, data, a["expected_tth"].as_str())
235    }
236}
netget/server/dc_peer/codec.rs

netget/server/dc_peer/
codec.rs
