remote_fs/
sftp.rs

1use std::path::{Path, PathBuf};
2use std::time::{Duration, SystemTime, UNIX_EPOCH};
3
4use async_ssh2_tokio::client::{AuthMethod, Client as SshClient, ServerCheckMethod};
5use async_trait::async_trait;
6use russh_sftp::client::SftpSession;
7use russh_sftp::protocol::FileType;
8use tokio::io::{AsyncReadExt, AsyncWriteExt};
9
10use crate::error::{Error, Result};
11use crate::traits::RemoteFileSystem;
12use crate::types::{FileEntry, FileMeta, SftpConfig};
13
14/// SFTP remote file system client (backed by `async-ssh2-tokio` + `russh-sftp`).
15pub struct SftpClient {
16    config: SftpConfig,
17    ssh: Option<SshClient>,
18    sftp: Option<SftpSession>,
19}
20
21impl SftpClient {
22    pub fn new(config: SftpConfig) -> Self {
23        Self {
24            config,
25            ssh: None,
26            sftp: None,
27        }
28    }
29
30    fn auth_method(&self) -> Result<AuthMethod> {
31        if let Some(ref password) = self.config.password {
32            Ok(AuthMethod::with_password(password))
33        } else if let Some(ref key) = self.config.private_key_path {
34            Ok(AuthMethod::with_key_file(key, None))
35        } else {
36            Err(Error::Auth(
37                "SFTP requires password or private key".into(),
38            ))
39        }
40    }
41
42    fn sftp(&self) -> Result<&SftpSession> {
43        self.sftp.as_ref().ok_or(Error::NotConnected)
44    }
45
46    async fn open_sftp(ssh: &SshClient) -> Result<SftpSession> {
47        let channel = ssh.get_channel().await?;
48        channel.request_subsystem(true, "sftp").await.map_err(|e| {
49            Error::Protocol(format!("failed to request sftp subsystem: {e}"))
50        })?;
51        let sftp = SftpSession::new(channel.into_stream()).await?;
52        Ok(sftp)
53    }
54}
55
56fn file_type_flags(ft: &FileType) -> (bool, bool, bool) {
57    match ft {
58        FileType::Dir => (true, false, false),
59        FileType::Symlink => (false, false, true),
60        FileType::File => (false, true, false),
61        _ => (false, true, false),
62    }
63}
64
65fn mtime_to_system(mtime: Option<u32>) -> Option<SystemTime> {
66    mtime.map(|secs| UNIX_EPOCH + Duration::from_secs(secs as u64))
67}
68
69#[async_trait]
70impl RemoteFileSystem for SftpClient {
71    async fn connect(&mut self) -> Result<()> {
72        let auth = self.auth_method()?;
73        let ssh = SshClient::connect(
74            (self.config.host.as_str(), self.config.port),
75            &self.config.username,
76            auth,
77            ServerCheckMethod::NoCheck,
78        )
79        .await?;
80        let sftp = Self::open_sftp(&ssh).await?;
81        self.ssh = Some(ssh);
82        self.sftp = Some(sftp);
83        Ok(())
84    }
85
86    async fn disconnect(&mut self) -> Result<()> {
87        if let Some(sftp) = self.sftp.take() {
88            let _ = sftp.close().await;
89        }
90        self.ssh = None;
91        Ok(())
92    }
93
94    fn is_connected(&self) -> bool {
95        self.ssh.is_some() && self.sftp.is_some()
96    }
97
98    async fn list(&mut self, path: &str) -> Result<Vec<FileEntry>> {
99        let sftp = self.sftp()?;
100        let mut entries = Vec::new();
101        let mut dir = sftp.read_dir(path.to_string()).await?;
102        while let Some(entry) = dir.next() {
103            let name = entry.file_name();
104            if name == "." || name == ".." {
105                continue;
106            }
107            let meta = entry.metadata();
108            let (is_dir, is_file, is_symlink) = file_type_flags(&meta.file_type());
109            let child = if path.ends_with('/') {
110                format!("{path}{name}")
111            } else if path.is_empty() {
112                name.clone()
113            } else {
114                format!("{path}/{name}")
115            };
116            entries.push(FileEntry {
117                path: PathBuf::from(child),
118                name,
119                is_dir,
120                is_file,
121                is_symlink,
122                size: meta.size.unwrap_or(0),
123                modified: mtime_to_system(meta.mtime),
124            });
125        }
126        Ok(entries)
127    }
128
129    async fn metadata(&mut self, path: &str) -> Result<FileMeta> {
130        let sftp = self.sftp()?;
131        let meta = sftp.metadata(path.to_string()).await?;
132        let (is_dir, is_file, is_symlink) = file_type_flags(&meta.file_type());
133        Ok(FileMeta {
134            path: PathBuf::from(path),
135            is_dir,
136            is_file,
137            is_symlink,
138            size: meta.size.unwrap_or(0),
139            modified: mtime_to_system(meta.mtime),
140        })
141    }
142
143    async fn exists(&mut self, path: &str) -> Result<bool> {
144        Ok(self.sftp()?.try_exists(path.to_string()).await?)
145    }
146
147    async fn mkdir(&mut self, path: &str) -> Result<()> {
148        self.sftp()?.create_dir(path.to_string()).await?;
149        Ok(())
150    }
151
152    async fn mkdir_all(&mut self, path: &str) -> Result<()> {
153        let path = path.trim_matches('/');
154        if path.is_empty() {
155            return Ok(());
156        }
157        let mut current = String::new();
158        for part in path.split('/').filter(|p| !p.is_empty()) {
159            current = if current.is_empty() {
160                format!("/{part}")
161            } else {
162                format!("{current}/{part}")
163            };
164            if !self.exists(&current).await? {
165                self.mkdir(&current).await?;
166            }
167        }
168        Ok(())
169    }
170
171    async fn rmdir(&mut self, path: &str) -> Result<()> {
172        self.sftp()?.remove_dir(path.to_string()).await?;
173        Ok(())
174    }
175
176    async fn remove_file(&mut self, path: &str) -> Result<()> {
177        self.sftp()?.remove_file(path.to_string()).await?;
178        Ok(())
179    }
180
181    async fn remove(&mut self, path: &str) -> Result<()> {
182        let meta = self.metadata(path).await?;
183        if meta.is_dir {
184            let children = self.list(path).await?;
185            for child in children {
186                let child_path = child.path.to_string_lossy().to_string();
187                Box::pin(self.remove(&child_path)).await?;
188            }
189            self.rmdir(path).await
190        } else {
191            self.remove_file(path).await
192        }
193    }
194
195    async fn rename(&mut self, from: &str, to: &str) -> Result<()> {
196        self.sftp()?
197            .rename(from.to_string(), to.to_string())
198            .await?;
199        Ok(())
200    }
201
202    async fn read(&mut self, path: &str) -> Result<Vec<u8>> {
203        Ok(self.sftp()?.read(path.to_string()).await?)
204    }
205
206    async fn write(&mut self, path: &str, data: &[u8]) -> Result<()> {
207        self.sftp()?.write(path.to_string(), data).await?;
208        Ok(())
209    }
210
211    async fn download(&mut self, remote: &str, local: &Path) -> Result<()> {
212        let data = self.read(remote).await?;
213        tokio::fs::write(local, data).await?;
214        Ok(())
215    }
216
217    async fn upload(&mut self, local: &Path, remote: &str) -> Result<()> {
218        let data = tokio::fs::read(local).await?;
219        self.write(remote, &data).await
220    }
221}
222
223#[allow(dead_code)]
224async fn copy_async_read_to_vec<R: AsyncReadExt + Unpin>(mut reader: R) -> Result<Vec<u8>> {
225    let mut buf = Vec::new();
226    reader.read_to_end(&mut buf).await?;
227    Ok(buf)
228}
229
230#[allow(dead_code)]
231async fn copy_slice_to_async_write<W: AsyncWriteExt + Unpin>(
232    mut writer: W,
233    data: &[u8],
234) -> Result<()> {
235    writer.write_all(data).await?;
236    writer.flush().await?;
237    Ok(())
238}
remote_fs/sftp.rs

remote_fs/
sftp.rs
