ssh_browser/sftp/
mod.rs

1//! SFTP framing with the send and receive halves deliberately separate.
2//!
3//! Nothing here pairs a request with its reply. That is the entire point. A page
4//! needs N subresources, and issuing all N requests before reading any reply is
5//! what holds the remote round trips at O(1) instead of O(N). An API that pairs
6//! one call to one reply — what every convenient sftp wrapper offers — puts that
7//! invariant out of reach, so this layer does not provide one.
8//!
9//! [`Sftp`] is the sequential form, used by the measurement binary. For concurrent
10//! callers the halves are split apart and driven by tasks; see `crate::fs::sftp`.
11
12pub mod transport;
13pub mod wire;
14
15use anyhow::{Context, Result, bail, ensure};
16use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
17
18use wire::{Dec, Enc, INIT, VERSION};
19
20const MAX_FRAME: usize = 64 * 1024 * 1024;
21
22pub struct Reply {
23    pub id: u32,
24    pub kind: u8,
25    body: Vec<u8>,
26}
27
28impl Reply {
29    /// Body past the leading request id.
30    pub fn payload(&self) -> &[u8] {
31        &self.body[4..]
32    }
33}
34
35pub async fn write_frame<W: AsyncWrite + Unpin>(w: &mut W, kind: u8, payload: &[u8]) -> Result<()> {
36    let len = u32::try_from(payload.len() + 1).context("sftp request too large")?;
37    w.write_all(&len.to_be_bytes()).await?;
38    w.write_all(&[kind]).await?;
39    w.write_all(payload).await?;
40    Ok(())
41}
42
43pub async fn read_frame<R: AsyncRead + Unpin>(r: &mut R) -> Result<(u8, Vec<u8>)> {
44    let mut head = [0u8; 5];
45    r.read_exact(&mut head).await?;
46    let len = u32::from_be_bytes(head[..4].try_into().expect("fixed size")) as usize;
47    if len == 0 || len > MAX_FRAME {
48        bail!("implausible sftp frame length {len}");
49    }
50    let mut body = vec![0u8; len - 1];
51    r.read_exact(&mut body).await?;
52    Ok((head[4], body))
53}
54
55/// Write half. `queue` buffers; only `flush` reaches the wire, so one flush can
56/// carry an arbitrary number of requests.
57pub struct Tx<W> {
58    w: W,
59    next_id: u32,
60}
61
62impl<W: AsyncWrite + Unpin> Tx<W> {
63    pub fn alloc_id(&mut self) -> u32 {
64        let id = self.next_id;
65        self.next_id = self.next_id.wrapping_add(1).max(1);
66        id
67    }
68
69    pub async fn queue(&mut self, kind: u8, payload: &[u8]) -> Result<()> {
70        write_frame(&mut self.w, kind, payload).await
71    }
72
73    pub async fn flush(&mut self) -> Result<()> {
74        self.w.flush().await?;
75        Ok(())
76    }
77}
78
79/// Read half. Replies arrive in whatever order the server finishes them, which is
80/// why every reply carries the request id back.
81pub struct Rx<R> {
82    r: R,
83}
84
85impl<R: AsyncRead + Unpin> Rx<R> {
86    pub async fn recv(&mut self) -> Result<Reply> {
87        let (kind, body) = read_frame(&mut self.r).await?;
88        ensure!(body.len() >= 4, "reply type {kind} carries no request id");
89        let id = u32::from_be_bytes(body[..4].try_into().expect("length checked above"));
90        Ok(Reply { id, kind, body })
91    }
92}
93
94pub struct Sftp<W, R> {
95    tx: Tx<W>,
96    rx: Rx<R>,
97    version: u32,
98}
99
100impl<W: AsyncWrite + Unpin, R: AsyncRead + Unpin> Sftp<W, R> {
101    pub async fn handshake(w: W, r: R) -> Result<Self> {
102        let mut tx = Tx { w, next_id: 1 };
103        let mut rx = Rx { r };
104
105        write_frame(&mut tx.w, INIT, &Enc::new().u32(3).done()).await?;
106        tx.flush().await?;
107
108        let (kind, body) = read_frame(&mut rx.r).await?;
109        ensure!(kind == VERSION, "expected SSH_FXP_VERSION, got type {kind}");
110        let version = Dec::new(&body).u32().context("malformed SSH_FXP_VERSION")?;
111        ensure!(
112            version >= 3,
113            "remote speaks sftp v{version}, need v3 or later"
114        );
115
116        Ok(Self { tx, rx, version })
117    }
118
119    pub fn version(&self) -> u32 {
120        self.version
121    }
122
123    pub fn alloc_id(&mut self) -> u32 {
124        self.tx.alloc_id()
125    }
126
127    pub async fn queue(&mut self, kind: u8, payload: &[u8]) -> Result<()> {
128        self.tx.queue(kind, payload).await
129    }
130
131    pub async fn flush(&mut self) -> Result<()> {
132        self.tx.flush().await
133    }
134
135    pub async fn recv(&mut self) -> Result<Reply> {
136        self.rx.recv().await
137    }
138
139    /// Hand the halves to separate tasks so concurrent callers can share one stream.
140    pub fn into_halves(self) -> (Tx<W>, Rx<R>) {
141        (self.tx, self.rx)
142    }
143}
ssh_browser/sftp/mod.rs

ssh_browser/sftp/
mod.rs
