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, Verb};
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    /// Send one request.
70    ///
71    /// A `Verb`, not a byte. `write_frame` below still takes a byte because a reply is a byte
72    /// and the test server writes replies -- but nothing outside this module reaches it to
73    /// send a *request*, and `Verb` has six values. See `wire::Verb`.
74    pub async fn queue(&mut self, kind: Verb, payload: &[u8]) -> Result<()> {
75        write_frame(&mut self.w, kind.code(), payload).await
76    }
77
78    pub async fn flush(&mut self) -> Result<()> {
79        self.w.flush().await?;
80        Ok(())
81    }
82}
83
84/// Read half. Replies arrive in whatever order the server finishes them, which is
85/// why every reply carries the request id back.
86pub struct Rx<R> {
87    r: R,
88}
89
90impl<R: AsyncRead + Unpin> Rx<R> {
91    pub async fn recv(&mut self) -> Result<Reply> {
92        let (kind, body) = read_frame(&mut self.r).await?;
93        ensure!(body.len() >= 4, "reply type {kind} carries no request id");
94        let id = u32::from_be_bytes(body[..4].try_into().expect("length checked above"));
95        Ok(Reply { id, kind, body })
96    }
97}
98
99pub struct Sftp<W, R> {
100    tx: Tx<W>,
101    rx: Rx<R>,
102    version: u32,
103}
104
105impl<W: AsyncWrite + Unpin, R: AsyncRead + Unpin> Sftp<W, R> {
106    pub async fn handshake(w: W, r: R) -> Result<Self> {
107        let mut tx = Tx { w, next_id: 1 };
108        let mut rx = Rx { r };
109
110        write_frame(&mut tx.w, INIT, &Enc::new().u32(3).done()).await?;
111        tx.flush().await?;
112
113        let (kind, body) = read_frame(&mut rx.r).await?;
114        ensure!(kind == VERSION, "expected SSH_FXP_VERSION, got type {kind}");
115        let version = Dec::new(&body).u32().context("malformed SSH_FXP_VERSION")?;
116        ensure!(
117            version >= 3,
118            "remote speaks sftp v{version}, need v3 or later"
119        );
120
121        Ok(Self { tx, rx, version })
122    }
123
124    pub fn version(&self) -> u32 {
125        self.version
126    }
127
128    pub fn alloc_id(&mut self) -> u32 {
129        self.tx.alloc_id()
130    }
131
132    pub async fn queue(&mut self, kind: Verb, payload: &[u8]) -> Result<()> {
133        self.tx.queue(kind, payload).await
134    }
135
136    pub async fn flush(&mut self) -> Result<()> {
137        self.tx.flush().await
138    }
139
140    pub async fn recv(&mut self) -> Result<Reply> {
141        self.rx.recv().await
142    }
143
144    /// Hand the halves to separate tasks so concurrent callers can share one stream.
145    pub fn into_halves(self) -> (Tx<W>, Rx<R>) {
146        (self.tx, self.rx)
147    }
148}
ssh_browser/sftp/mod.rs

ssh_browser/sftp/
mod.rs
