netget/client/ssh/
sftp.rs

1//! Owned bounded SFTP v3 requests. No local filesystem or detached session tasks.
2//! Wire reference: draft-ietf-secsh-filexfer-02, the version used by OpenSSH.
3use anyhow::{bail, ensure, Context, Result};
4use serde_json::{json, Value};
5use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
6
7pub const MAX_PACKET: usize = 64 * 1024;
8pub const MAX_PATH: usize = 4096;
9pub const MAX_TEXT: usize = 1024 * 1024;
10pub const MAX_ENTRIES: usize = 1024;
11pub const MAX_CHUNK: usize = 32768;
12pub const DEFAULT_READ: usize = 65536;
13
14pub fn validate_action(action: &Value) -> Result<()> {
15    let path = action["path"].as_str().context("SFTP needs a path")?;
16    ensure!(
17        !path.is_empty() && path.len() <= MAX_PATH && !path.contains('\0'),
18        "SFTP path must contain 1..4096 bytes without NUL"
19    );
20    if let Some(offset) = action.get("offset") {
21        offset
22            .as_u64()
23            .context("offset must be an unsigned integer")?;
24    }
25    if let Some(length) = action.get("length") {
26        ensure!(
27            length
28                .as_u64()
29                .is_some_and(|n| n > 0 && n <= MAX_TEXT as u64),
30            "length must be 1..1048576"
31        );
32    }
33    if let Some(follow) = action.get("follow_symlinks") {
34        ensure!(follow.is_boolean(), "follow_symlinks must be a boolean");
35    }
36    ensure!(
37        action["offset"]
38            .as_u64()
39            .unwrap_or(0)
40            .checked_add(action["length"].as_u64().unwrap_or(DEFAULT_READ as u64))
41            .is_some(),
42        "SFTP offset overflow"
43    );
44    Ok(())
45}
46
47struct Cursor<'a>(&'a [u8]);
48impl<'a> Cursor<'a> {
49    fn take(&mut self, n: usize) -> Result<&'a [u8]> {
50        ensure!(n <= self.0.len(), "Truncated SFTP field");
51        let (value, rest) = self.0.split_at(n);
52        self.0 = rest;
53        Ok(value)
54    }
55    fn u32(&mut self) -> Result<u32> {
56        Ok(u32::from_be_bytes(self.take(4)?.try_into()?))
57    }
58    fn u64(&mut self) -> Result<u64> {
59        Ok(u64::from_be_bytes(self.take(8)?.try_into()?))
60    }
61    fn string(&mut self, cap: usize) -> Result<&'a [u8]> {
62        let n = self.u32()? as usize;
63        ensure!(n <= cap, "SFTP string exceeds bound");
64        self.take(n)
65    }
66    fn text(&mut self) -> Result<String> {
67        Ok(std::str::from_utf8(self.string(MAX_PATH)?)?.into())
68    }
69    fn attrs(&mut self) -> Result<Value> {
70        let flags = self.u32()?;
71        ensure!(flags & !0x8000000f == 0, "Unknown SFTP attribute flags");
72        let mut a = serde_json::Map::new();
73        if flags & 1 != 0 {
74            a.insert("size".into(), json!(self.u64()?));
75        }
76        if flags & 2 != 0 {
77            a.insert("uid".into(), json!(self.u32()?));
78            a.insert("gid".into(), json!(self.u32()?));
79        }
80        if flags & 4 != 0 {
81            let mode = self.u32()?;
82            a.insert("permissions".into(), json!(mode));
83            a.insert("is_dir".into(), json!(mode & 0o170000 == 0o040000));
84            a.insert("is_symlink".into(), json!(mode & 0o170000 == 0o120000));
85        }
86        if flags & 8 != 0 {
87            a.insert("atime".into(), json!(self.u32()?));
88            a.insert("mtime".into(), json!(self.u32()?));
89        }
90        if flags & 0x80000000 != 0 {
91            let n = self.u32()? as usize;
92            ensure!(
93                n <= 64 && n <= self.0.len() / 8,
94                "SFTP extended attribute count exceeds bound"
95            );
96            // Attribute extensions are opaque protocol metadata, never model bytes.
97            for _ in 0..n {
98                self.string(MAX_PATH)?;
99                self.string(MAX_PATH)?;
100            }
101        }
102        Ok(Value::Object(a))
103    }
104    fn finish(self) -> Result<()> {
105        ensure!(self.0.is_empty(), "Trailing SFTP packet bytes");
106        Ok(())
107    }
108}
109
110#[derive(Debug)]
111pub enum Reply {
112    Status { code: u32, message: String },
113    Handle(Vec<u8>),
114    Data(Vec<u8>),
115    Names(Vec<Value>),
116    Attributes(Value),
117}
118/// Counts and strings are validated before allocating collections or copying data.
119pub fn decode_reply(bytes: &[u8], expected_id: u32) -> Result<Reply> {
120    ensure!(bytes.len() <= MAX_PACKET, "SFTP packet exceeds 64 KiB");
121    let mut c = Cursor(bytes);
122    let kind = c.take(1)?[0];
123    ensure!(c.u32()? == expected_id, "Mismatched SFTP response ID");
124    let reply = match kind {
125        101 => {
126            let code = c.u32()?;
127            ensure!(code <= 8, "Unknown SFTP status");
128            let message = c.text()?;
129            c.string(MAX_PATH)?;
130            Reply::Status { code, message }
131        }
132        102 => {
133            let handle = c.string(256)?;
134            ensure!(!handle.is_empty(), "Empty SFTP handle");
135            Reply::Handle(handle.to_vec())
136        }
137        103 => Reply::Data(c.string(MAX_CHUNK)?.to_vec()),
138        104 => {
139            let n = c.u32()? as usize;
140            ensure!(
141                n <= MAX_ENTRIES && n <= c.0.len() / 12,
142                "SFTP directory entry count exceeds bound"
143            );
144            let mut entries = Vec::with_capacity(n);
145            for _ in 0..n {
146                let name = c.text()?;
147                c.string(MAX_PATH)?; // Human-readable longname is not a typed attribute.
148                entries.push(json!({"name":name,"attributes":c.attrs()?}));
149            }
150            Reply::Names(entries)
151        }
152        105 => Reply::Attributes(c.attrs()?),
153        _ => bail!("Unexpected SFTP packet type"),
154    };
155    c.finish()?;
156    Ok(reply)
157}
158
159pub async fn read_packet<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Vec<u8>> {
160    let length = reader.read_u32().await? as usize;
161    ensure!(
162        (1..=MAX_PACKET).contains(&length),
163        "SFTP packet length exceeds 64 KiB"
164    );
165    let mut bytes = vec![0; length];
166    reader.read_exact(&mut bytes).await?;
167    Ok(bytes)
168}
169fn string(bytes: &[u8]) -> Vec<u8> {
170    let mut out = (bytes.len() as u32).to_be_bytes().to_vec();
171    out.extend_from_slice(bytes);
172    out
173}
174struct RawSession<S> {
175    stream: S,
176    id: u32,
177    received: usize,
178}
179impl<S: AsyncRead + AsyncWrite + Unpin> RawSession<S> {
180    async fn receive(&mut self) -> Result<Vec<u8>> {
181        let bytes = read_packet(&mut self.stream).await?;
182        self.received += bytes.len();
183        ensure!(
184            self.received <= 2 * MAX_TEXT,
185            "SFTP operation packet budget exceeded"
186        );
187        Ok(bytes)
188    }
189    async fn send(&mut self, bytes: &[u8]) -> Result<()> {
190        ensure!(
191            bytes.len() <= MAX_PACKET,
192            "SFTP outgoing packet exceeds bound"
193        );
194        self.stream.write_u32(bytes.len() as u32).await?;
195        self.stream.write_all(bytes).await?;
196        self.stream.flush().await?;
197        Ok(())
198    }
199    async fn init(&mut self) -> Result<()> {
200        self.send(&[1, 0, 0, 0, 3]).await?;
201        let bytes = self.receive().await?;
202        let mut c = Cursor(&bytes);
203        ensure!(
204            c.take(1)?[0] == 2 && c.u32()? == 3,
205            "SFTP peer must negotiate version 3"
206        );
207        let mut extensions = 0;
208        while !c.0.is_empty() {
209            extensions += 1;
210            ensure!(extensions <= 64, "Too many SFTP extensions");
211            c.string(MAX_PATH)?;
212            c.string(MAX_PATH)?;
213        }
214        Ok(())
215    }
216    async fn request(&mut self, kind: u8, payload: &[u8]) -> Result<Reply> {
217        self.id = self
218            .id
219            .checked_add(1)
220            .context("SFTP request ID exhausted")?;
221        let mut out = vec![kind];
222        out.extend_from_slice(&self.id.to_be_bytes());
223        out.extend_from_slice(payload);
224        self.send(&out).await?;
225        decode_reply(&self.receive().await?, self.id)
226    }
227    async fn close(&mut self, handle: &[u8]) -> Result<()> {
228        match self.request(4, &string(handle)).await? {
229            Reply::Status { code: 0, .. } => Ok(()),
230            _ => bail!("SFTP handle close failed"),
231        }
232    }
233}
234fn status(reply: Reply) -> Result<Reply> {
235    match reply {
236        Reply::Status { code, message } => bail!("SFTP status {code}: {message}"),
237        reply => Ok(reply),
238    }
239}
240/// Each operation gets its own channel; handles stay inside the exchange and are
241/// closed on success. The owner closes the channel on all other outcomes.
242pub async fn exchange<S: AsyncRead + AsyncWrite + Unpin>(
243    stream: S,
244    action: &Value,
245) -> Result<Value> {
246    validate_action(action)?;
247    let mut s = RawSession {
248        stream,
249        id: 0,
250        received: 0,
251    };
252    s.init().await?;
253    let path = action["path"].as_str().context("Missing path")?;
254    match action["type"].as_str().unwrap_or("") {
255        "sftp_stat" => {
256            let kind = if action["follow_symlinks"].as_bool().unwrap_or(false) {
257                17
258            } else {
259                7
260            };
261            match status(s.request(kind, &string(path.as_bytes())).await?)? {
262                Reply::Attributes(attrs) => {
263                    Ok(json!({"operation":"stat","path":path,"attributes":attrs}))
264                }
265                _ => bail!("SFTP stat expected attributes"),
266            }
267        }
268        "sftp_list_directory" => {
269            let handle = match status(s.request(11, &string(path.as_bytes())).await?)? {
270                Reply::Handle(h) => h,
271                _ => bail!("SFTP opendir expected handle"),
272            };
273            let mut entries = Vec::new();
274            let mut empty_batches = 0;
275            loop {
276                match s.request(12, &string(&handle)).await? {
277                    Reply::Status { code: 1, .. } => break,
278                    Reply::Names(batch) => {
279                        // A peer can return an empty NAME before the final EOF,
280                        // including NetGet's semantic empty-directory response.
281                        if batch.is_empty() {
282                            empty_batches += 1;
283                            ensure!(empty_batches <= 16, "SFTP directory made no progress");
284                        }
285                        ensure!(
286                            batch.len() <= MAX_ENTRIES.saturating_sub(entries.len()),
287                            "SFTP directory exceeds 1024 entries"
288                        );
289                        entries.extend(batch);
290                    }
291                    reply => {
292                        status(reply)?;
293                        bail!("SFTP readdir expected names");
294                    }
295                }
296            }
297            s.close(&handle).await?;
298            Ok(json!({"operation":"list_directory","path":path,"entries":entries}))
299        }
300        "sftp_read_file" => {
301            let mut payload = string(path.as_bytes());
302            payload.extend_from_slice(&1u32.to_be_bytes()); // SSH_FXF_READ
303            payload.extend_from_slice(&0u32.to_be_bytes()); // no invented attributes
304            let handle = match status(s.request(3, &payload).await?)? {
305                Reply::Handle(h) => h,
306                _ => bail!("SFTP open expected handle"),
307            };
308            let offset = action["offset"].as_u64().unwrap_or(0);
309            let length = action["length"].as_u64().unwrap_or(DEFAULT_READ as u64) as usize;
310            let mut data = Vec::new();
311            let mut eof = false;
312            while data.len() < length {
313                let count = MAX_CHUNK.min(length - data.len());
314                let mut payload = string(&handle);
315                payload.extend_from_slice(&(offset + data.len() as u64).to_be_bytes());
316                payload.extend_from_slice(&(count as u32).to_be_bytes());
317                match s.request(5, &payload).await? {
318                    Reply::Status { code: 1, .. } => {
319                        eof = true;
320                        break;
321                    }
322                    Reply::Data(chunk) => {
323                        ensure!(
324                            !chunk.is_empty() && chunk.len() <= count,
325                            "SFTP read returned an invalid data length"
326                        );
327                        data.extend(chunk);
328                    }
329                    reply => {
330                        status(reply)?;
331                        bail!("SFTP read expected data");
332                    }
333                }
334            }
335            s.close(&handle).await?;
336            let bytes = data.len();
337            let text = String::from_utf8(data).context("SFTP file window is not UTF-8 text")?;
338            Ok(
339                json!({"operation":"read_file","path":path,"offset":offset,"bytes_read":bytes,"eof":eof,"text":text}),
340            )
341        }
342        _ => bail!("Unknown SFTP operation"),
343    }
344}
netget/client/ssh/sftp.rs

netget/client/ssh/
sftp.rs
