netget/server/adc_peer/
codec.rs

1//! Direct ADC/ADCS peer file transfers; content belongs to request handlers.
2use crate::server::{
3    adc::codec::{cid, encode, hash, words},
4    p2p_support::{
5        files::{content, decoded_result, MAX_FILE},
6        DeviceSession, Framer, ReadStream, ScannerSession, Stream,
7    },
8};
9use anyhow::{bail, ensure, Context, Result};
10use async_trait::async_trait;
11use serde_json::{json, Value};
12use tokio::io::{AsyncReadExt, AsyncWriteExt};
13pub const MAX_COMMAND: usize = 64 * 1024;
14pub fn identity() -> String {
15    use rand::RngCore;
16    let mut random = [0u8; 24];
17    rand::thread_rng().fill_bytes(&mut random);
18    hash(&random)
19}
20pub fn validate(v: &Value) -> Result<()> {
21    ensure!(v["type"] == "adc_peer_get", "unknown ADC peer action");
22    let id = v["identifier"].as_str().context("identifier")?;
23    ensure!(
24        !id.is_empty() && id.len() <= 4096 && !id.contains(['\0', '\r', '\n']),
25        "identifier"
26    );
27    ensure!(v["offset"].as_u64().is_some(), "offset");
28    let len = v["length"].as_i64().context("length")?;
29    ensure!(
30        len == -1 || (0..=MAX_FILE as i64).contains(&len),
31        "transfer too large"
32    );
33    Ok(())
34}
35pub struct Device {
36    frame: Framer,
37    supported: bool,
38    identified: bool,
39    cid: String,
40}
41impl Device {
42    pub fn with_cid(cid: String) -> Self {
43        Self {
44            frame: Framer::default(),
45            supported: false,
46            identified: false,
47            cid,
48        }
49    }
50}
51impl Default for Device {
52    fn default() -> Self {
53        Self {
54            frame: Framer::default(),
55            supported: false,
56            identified: false,
57            cid: identity(),
58        }
59    }
60}
61#[async_trait]
62impl DeviceSession for Device {
63    async fn read(&mut self, s: &mut ReadStream) -> Result<Vec<u8>> {
64        self.frame.delimited(s, b'\n', MAX_COMMAND).await
65    }
66    fn receive(&mut self, b: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
67        let w = words(b)?;
68        if w[0] == "CSUP" {
69            ensure!(
70                !self.supported
71                    && w.iter().any(|v| v == "ADBASE")
72                    && w.iter().any(|v| v == "ADTIGR"),
73                "ADC BASE/TIGR required"
74            );
75            self.supported = true;
76            return Ok((
77                format!("CSUP ADBASE ADTIGR\nCINF ID{}\n", self.cid).into_bytes(),
78                None,
79            ));
80        }
81        if w[0] == "CSTA" {
82            let code = w.get(1).context("status code")?;
83            ensure!(
84                code.len() == 3 && code.bytes().all(|b| b.is_ascii_digit()),
85                "invalid status"
86            );
87            ensure!(!code.starts_with('2'), "peer fatal status");
88            return Ok((vec![], None));
89        }
90        if w[0] == "CINF" {
91            ensure!(self.supported && !self.identified, "peer negotiation order");
92            cid(w
93                .iter()
94                .skip(1)
95                .find_map(|w| w.strip_prefix("ID"))
96                .context("CID required")?)?;
97            self.identified = true;
98            return Ok((vec![], None));
99        }
100        ensure!(
101            self.supported && self.identified,
102            "peer negotiation incomplete"
103        );
104        if w[0] != "CGET" {
105            return Ok((b"CSTA 140 Unsupported\n".to_vec(), None));
106        }
107        ensure!(w.len() == 5 && w[1] == "file", "selected file request only");
108        let offset: u64 = w[3].parse()?;
109        let length: i64 = w[4].parse()?;
110        let v = json!({"type":"adc_peer_get","operation":"get","identifier":w[2],"offset":offset,"length":length});
111        validate(&v)?;
112        Ok((vec![], Some(v)))
113    }
114    fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
115        match content(r, a) {
116            Ok(data) => {
117                let mut bytes = encode(
118                    "CSND",
119                    &[
120                        "file".into(),
121                        r["identifier"].as_str().context("identifier")?.into(),
122                        r["offset"].to_string(),
123                        data.len().to_string(),
124                    ],
125                )?;
126                bytes.extend(data);
127                Ok(bytes)
128            }
129            Err(_) => Ok(b"CSTA 151 File\\snot\\savailable\n".to_vec()),
130        }
131    }
132}
133pub struct Scanner {
134    frame: Framer,
135    cid: String,
136    token: Option<String>,
137}
138impl Scanner {
139    pub fn with_identity(id: Option<String>, token: Option<String>) -> Result<Self> {
140        let id = id.unwrap_or_else(identity);
141        cid(&id)?;
142        if let Some(t) = &token {
143            ensure!(
144                !t.is_empty() && t.len() <= 128 && !t.contains([' ', '\n', '\r', '\0']),
145                "invalid token"
146            );
147        }
148        Ok(Self {
149            frame: Framer::default(),
150            cid: id,
151            token,
152        })
153    }
154}
155impl Default for Scanner {
156    fn default() -> Self {
157        Self {
158            frame: Framer::default(),
159            cid: identity(),
160            token: None,
161        }
162    }
163}
164#[async_trait]
165impl ScannerSession for Scanner {
166    async fn open(&mut self, s: &mut Stream) -> Result<()> {
167        s.write_all(
168            format!(
169                "CSUP ADBASE ADTIGR\nCINF ID{}{}\n",
170                self.cid,
171                self.token
172                    .as_ref()
173                    .map_or(String::new(), |t| format!(" TO{t}"))
174            )
175            .as_bytes(),
176        )
177        .await?;
178        let mut supported = false;
179        for _ in 0..16 {
180            let w = words(&self.frame.delimited(s, b'\n', MAX_COMMAND).await?)?;
181            match w[0].as_str() {
182                "CSUP" => {
183                    ensure!(
184                        w.iter().any(|v| v == "ADBASE") && w.iter().any(|v| v == "ADTIGR"),
185                        "peer capabilities"
186                    );
187                    supported = true;
188                }
189                "CINF" => {
190                    ensure!(supported, "peer order");
191                    cid(w
192                        .iter()
193                        .skip(1)
194                        .find_map(|w| w.strip_prefix("ID"))
195                        .context("CID required")?)?;
196                    return Ok(());
197                }
198                "CSTA" => {
199                    let code = w.get(1).context("status code")?;
200                    ensure!(!code.starts_with('2'), "peer refused connection");
201                }
202                _ => {}
203            }
204        }
205        bail!("peer handshake exceeded cap")
206    }
207    async fn idle(&mut self, s: &mut Stream) -> Result<Option<Value>> {
208        let b = self.frame.delimited(s, b'\n', MAX_COMMAND).await?;
209        if b == b"\n" {
210            return Ok(None);
211        }
212        let w = words(&b)?;
213        ensure!(
214            w[0] == "CSTA" && w.get(1).is_some_and(|c| c.starts_with('0')),
215            "unexpected peer frame"
216        );
217        Ok(None)
218    }
219    async fn exchange(&mut self, s: &mut Stream, a: &Value) -> Result<Value> {
220        validate(a)?;
221        let id = a["identifier"].as_str().context("identifier")?;
222        s.write_all(&encode(
223            "CGET",
224            &[
225                "file".into(),
226                id.into(),
227                a["offset"].to_string(),
228                a["length"].to_string(),
229            ],
230        )?)
231        .await?;
232        let w = words(&self.frame.delimited(s, b'\n', MAX_COMMAND).await?)?;
233        if w[0] == "CSTA" {
234            return Ok(json!({"error":w.get(1),"message":w.get(2)}));
235        }
236        ensure!(
237            w.len() == 5 && w[0] == "CSND" && w[1] == "file" && w[2] == id,
238            "transfer response"
239        );
240        ensure!(
241            w[3].parse::<u64>()? == a["offset"].as_u64().context("offset")?,
242            "offset mismatch"
243        );
244        let count: usize = w[4].parse()?;
245        ensure!(count <= MAX_FILE, "transfer exceeds cap");
246        if let Some(length) = a["length"].as_i64().filter(|n| *n >= 0) {
247            ensure!(count == length as usize, "length mismatch");
248        }
249        let mut data = vec![0; count];
250        s.read_exact(&mut data).await?;
251        decoded_result(id, data, a["expected_tth"].as_str())
252    }
253}
netget/server/adc_peer/codec.rs

netget/server/adc_peer/
codec.rs
