netget/server/soulseek_peer/
codec.rs

1//! Soulseek P-connection browsing; compressed content is bounded after decompression.
2use crate::server::{
3    p2p_support::{DeviceSession, Framer, ReadStream, ScannerSession, Stream},
4    soulseek::codec::{string, Reader},
5};
6use anyhow::{bail, ensure, Context, Result};
7use async_trait::async_trait;
8use serde_json::{json, Value};
9use std::io::{Read, Write};
10use tokio::io::AsyncWriteExt;
11pub const MAX_COMMAND: usize = 1024 * 1024;
12fn length(b: &[u8]) -> Result<usize> {
13    let n = u32::from_le_bytes(b.try_into()?) as usize;
14    ensure!((1..=MAX_COMMAND - 4).contains(&n), "peer frame length");
15    Ok(n + 4)
16}
17fn packet(code: u32, data: Vec<u8>) -> Result<Vec<u8>> {
18    ensure!(data.len() <= MAX_COMMAND - 8, "peer packet too large");
19    let mut b = ((data.len() + 4) as u32).to_le_bytes().to_vec();
20    b.extend(code.to_le_bytes());
21    b.extend(data);
22    Ok(b)
23}
24pub fn validate(v: &Value) -> Result<()> {
25    ensure!(
26        matches!(
27            v["type"].as_str(),
28            Some("soulseek_peer_shares" | "soulseek_peer_info")
29        ),
30        "unknown peer action"
31    );
32    Ok(())
33}
34fn directories(out: &mut Vec<u8>, v: &Value) -> Result<()> {
35    let dirs = v.as_array().cloned().unwrap_or_default();
36    ensure!(dirs.len() <= 128, "directory count");
37    out.extend((dirs.len() as u32).to_le_bytes());
38    for d in dirs {
39        string(out, d["name"].as_str().context("directory name")?)?;
40        let files = d["files"].as_array().context("files")?;
41        ensure!(files.len() <= 128, "file count");
42        out.extend((files.len() as u32).to_le_bytes());
43        for f in files {
44            out.push(1);
45            string(out, f["name"].as_str().context("filename")?)?;
46            out.extend(f["size"].as_u64().context("size")?.to_le_bytes());
47            string(out, f["extension"].as_str().unwrap_or(""))?;
48            out.extend(0u32.to_le_bytes());
49            ensure!(out.len() <= MAX_COMMAND, "listing too large");
50        }
51    }
52    Ok(())
53}
54fn read_dirs(r: &mut Reader<'_>) -> Result<Value> {
55    let count = r.u32()?;
56    ensure!(count <= 128, "directory count");
57    let mut dirs = vec![];
58    for _ in 0..count {
59        let name = r.string()?;
60        let n = r.u32()?;
61        ensure!(n <= 128, "file count");
62        let mut files = vec![];
63        for _ in 0..n {
64            ensure!(r.take(1)?[0] == 1, "file marker");
65            let name = r.string()?;
66            let size = r.u64()?;
67            let extension = r.string()?;
68            let n = r.u32()?;
69            ensure!(n <= 32, "attribute count");
70            let mut attributes = vec![];
71            for _ in 0..n {
72                attributes.push(json!({"id":r.u32()?,"value":r.u32()?}));
73            }
74            files.push(
75                json!({"name":name,"size":size,"extension":extension,"attributes":attributes}),
76            );
77        }
78        dirs.push(json!({"name":name,"files":files}));
79    }
80    Ok(json!(dirs))
81}
82#[derive(Default)]
83pub struct Device {
84    frame: Framer,
85    initialized: bool,
86}
87#[async_trait]
88impl DeviceSession for Device {
89    async fn read(&mut self, s: &mut ReadStream) -> Result<Vec<u8>> {
90        self.frame.sized(s, 4, MAX_COMMAND, length).await
91    }
92    fn receive(&mut self, b: &[u8]) -> Result<(Vec<u8>, Option<Value>)> {
93        let mut r = Reader::new(&b[4..]);
94        if !self.initialized {
95            ensure!(r.take(1)?[0] == 1, "PeerInit required");
96            let name = r.string()?;
97            ensure!(!name.is_empty(), "peer username");
98            ensure!(r.string()? == "P", "selected P connection only");
99            r.u32()?;
100            if r.pos < r.bytes.len() {
101                r.u32()?;
102            }
103            r.end()?;
104            self.initialized = true;
105            return Ok((vec![], None));
106        }
107        let code = r.u32()?;
108        match code {
109            4 => {
110                if r.pos < r.bytes.len() {
111                    r.u32()?;
112                }
113            }
114            15 => {}
115            _ => return Ok((vec![], None)),
116        }
117        r.end()?;
118        Ok((
119            vec![],
120            Some(json!({"operation":if code==4{"shares"}else{"info"},"code":code})),
121        ))
122    }
123    fn answer(&mut self, r: &Value, a: Option<&Value>) -> Result<Vec<u8>> {
124        let empty = json!({});
125        let a = a.unwrap_or(&empty);
126        let mut out = vec![];
127        match r["code"].as_u64().context("code")? {
128            4 => {
129                directories(&mut out, &a["directories"])?;
130                out.extend([0; 8]);
131                ensure!(out.len() <= MAX_COMMAND, "decoded listing cap");
132                let mut enc =
133                    flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
134                enc.write_all(&out)?;
135                packet(5, enc.finish()?)
136            }
137            15 => {
138                string(&mut out, a["description"].as_str().unwrap_or(""))?;
139                out.push(0);
140                for key in ["upload_slots", "queue_size"] {
141                    let n = a[key].as_u64().unwrap_or(0);
142                    ensure!(n <= u32::MAX as u64, "{key}");
143                    out.extend((n as u32).to_le_bytes());
144                }
145                out.push(u8::from(a["has_slots_free"] == true));
146                out.extend(0u32.to_le_bytes());
147                packet(16, out)
148            }
149            _ => bail!("unknown peer reply"),
150        }
151    }
152}
153fn response(b: &[u8]) -> Result<Value> {
154    let mut r = Reader::new(&b[4..]);
155    let code = r.u32()?;
156    match code {
157        5 => {
158            let mut out = vec![];
159            flate2::read::ZlibDecoder::new(&r.bytes[r.pos..])
160                .take((MAX_COMMAND + 1) as u64)
161                .read_to_end(&mut out)?;
162            ensure!(out.len() <= MAX_COMMAND, "decompression cap");
163            let mut r = Reader::new(&out);
164            let dirs = read_dirs(&mut r)?;
165            if r.pos < r.bytes.len() {
166                r.u32()?;
167            }
168            let locked = if r.pos < r.bytes.len() {
169                read_dirs(&mut r)?
170            } else {
171                json!([])
172            };
173            r.end()?;
174            Ok(json!({"code":code,"directories":dirs,"locked_directories":locked}))
175        }
176        16 => {
177            let description = r.string()?;
178            let picture = r.boolean()?;
179            ensure!(!picture, "pictures outside selected scope");
180            let slots = r.u32()?;
181            let queue = r.u32()?;
182            let free = r.boolean()?;
183            let permissions = if r.pos < r.bytes.len() { r.u32()? } else { 0 };
184            ensure!(permissions <= 3, "permissions");
185            r.end()?;
186            Ok(
187                json!({"code":code,"description":description,"upload_slots":slots,"queue_size":queue,"has_slots_free":free,"upload_permissions":permissions}),
188            )
189        }
190        _ => bail!("unsupported peer reply"),
191    }
192}
193#[derive(Default)]
194pub struct Scanner {
195    frame: Framer,
196}
197#[async_trait]
198impl ScannerSession for Scanner {
199    async fn open(&mut self, s: &mut Stream) -> Result<()> {
200        let mut init = vec![1];
201        string(&mut init, "NetGet")?;
202        string(&mut init, "P")?;
203        init.extend(0u32.to_le_bytes());
204        let mut b = (init.len() as u32).to_le_bytes().to_vec();
205        b.extend(init);
206        s.write_all(&b).await?;
207        Ok(())
208    }
209    async fn exchange(&mut self, s: &mut Stream, a: &Value) -> Result<Value> {
210        validate(a)?;
211        let code = if a["type"] == "soulseek_peer_shares" {
212            4
213        } else {
214            15
215        };
216        s.write_all(&packet(code, vec![])?).await?;
217        let v = response(&self.frame.sized(s, 4, MAX_COMMAND, length).await?)?;
218        ensure!(
219            v["code"] == if code == 4 { 5 } else { 16 },
220            "mismatched peer reply"
221        );
222        Ok(v)
223    }
224    async fn idle(&mut self, s: &mut Stream) -> Result<Option<Value>> {
225        Ok(Some(response(
226            &self.frame.sized(s, 4, MAX_COMMAND, length).await?,
227        )?))
228    }
229}
netget/server/soulseek_peer/codec.rs

netget/server/soulseek_peer/
codec.rs
