CH-J Toolsupravljanje poslužiteljima preko SSH
Meni
Objavljen izvor

CH-J Tools

Pregledaj direktorije i datoteke za određeni programski izdanje.

Preuzmi ZIP izvora
CH-J / Server Manager 1704 datoteke
0.1.2-beta Dodano: · Izgrađeno:
sftp.rs Dodano: Datum nije zabilježen 12,7 KB · 321 linije
Oznaka Preuzmi datoteku
1//! SFTP channels over the shared SSH connection (replaces ssh2 `client.sftp()`).
2//!
3//! Built on `russh_sftp::client::RawSftpSession` so positioned reads/writes can be pipelined
4//! (the Transfer Manager keeps up to 8 requests in flight, like 0.1.0). Errors keep the SFTP
5//! status code so callers can map them exactly like the 0.1.0 `sftpError`/`errorCode` helpers.
7use crate::core::error::{CoreError, CoreResult};
8use crate::core::ssh::session::{not_connected, SessionManager, SessionRecord};
9use russh_sftp::client::error::Error as RawError;
10use russh_sftp::client::RawSftpSession;
11use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode};
12use std::sync::Arc;
13use std::time::Duration;
15/// SFTP request timeout (a dead link never answers; the transfer stall detector uses 60 s too).
16const REQUEST_TIMEOUT_SECS: u64 = 60;
18#[derive(Debug, Clone)]
19pub enum SftpError {
20 /// SSH_FX_* status code and server message.
21 Status(u32, String),
22 /// Channel closed / connection lost / timeout.
23 Connection(String),
24 Other(String),
27impl SftpError {
28 pub fn status(&self) -> Option<u32> {
29 match self {
30 SftpError::Status(code, _) => Some(*code),
31 _ => None,
32 }
33 }
35 pub fn message(&self) -> String {
36 match self {
37 SftpError::Status(_, m) | SftpError::Connection(m) | SftpError::Other(m) => m.clone(),
38 }
39 }
41 /// 0.1.0 `sftpError`: FILE_* codes for the File Manager.
42 pub fn file_error(&self, message: &str) -> CoreError {
43 let code = match self {
44 SftpError::Status(2, _) => "FILE_NOT_FOUND",
45 SftpError::Status(3, _) => "FILE_PERMISSION_DENIED",
46 SftpError::Status(6 | 7, _) | SftpError::Connection(_) => "SESSION_NOT_CONNECTED",
47 SftpError::Status(8, _) => "FILE_OPERATION_UNSUPPORTED",
48 _ => "FILE_REMOTE_IO_FAILED",
49 };
50 let text = if message.is_empty() { self.message() } else { message.to_string() };
51 CoreError::new(code, format!("{code}: {text}"))
52 }
54 /// 0.1.0 Transfer Manager `errorCode`.
55 pub fn transfer_code(&self) -> &'static str {
56 match self {
57 SftpError::Status(6 | 7, _) | SftpError::Connection(_) => "SESSION_NOT_CONNECTED",
58 SftpError::Status(3, _) => "FILE_PERMISSION_DENIED",
59 SftpError::Status(2, _) => "FILE_NOT_FOUND",
60 _ => "TRANSFER_FAILED",
61 }
62 }
65impl std::fmt::Display for SftpError {
66 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67 f.write_str(&self.message())
68 }
71fn convert(error: RawError) -> SftpError {
72 match error {
73 RawError::Status(status) => SftpError::Status(status_number(status.status_code), status.error_message),
74 RawError::Timeout => SftpError::Connection("SFTP request timed out.".into()),
75 RawError::UnexpectedBehavior(m) if m.contains("closed") => SftpError::Connection(m),
76 RawError::IO(m) => SftpError::Connection(m),
77 other => SftpError::Other(other.to_string()),
78 }
81fn status_number(code: StatusCode) -> u32 {
82 code as u32
85pub type SftpResult<T> = Result<T, SftpError>;
87/// Normalized attributes (ssh2 `Stats` equivalent).
88#[derive(Debug, Clone, Default)]
89pub struct Stat {
90 pub size: u64,
91 pub mode: u32,
92 pub mtime: u32,
93 pub uid: Option<u32>,
94 pub gid: Option<u32>,
97const S_IFMT: u32 = 0o170000;
99impl Stat {
100 fn from(attrs: &FileAttributes) -> Stat {
101 Stat { size: attrs.size.unwrap_or(0), mode: attrs.permissions.unwrap_or(0), mtime: attrs.mtime.unwrap_or(0), uid: attrs.uid, gid: attrs.gid }
102 }
103 pub fn is_dir(&self) -> bool {
104 self.mode & S_IFMT == 0o040000
105 }
106 pub fn is_file(&self) -> bool {
107 self.mode & S_IFMT == 0o100000
108 }
109 pub fn is_symlink(&self) -> bool {
110 self.mode & S_IFMT == 0o120000
111 }
114#[derive(Debug, Clone)]
115pub struct DirEntry {
116 pub name: String,
117 pub stat: Stat,
120/// One SFTP subsystem channel. Cheap to share (`Arc`); requests may run concurrently.
121pub struct Sftp {
122 raw: RawSftpSession,
123 /// Largest data payload per READ/WRITE request (ssh2 rule: OpenSSH 256 KiB packets,
124 /// otherwise 34000 bytes; `limits@openssh.com` wins when offered).
125 max_read: usize,
126 max_write: usize,
129/// Open file handle (closed explicitly; dropping without close leaks only the server handle
130/// until the channel closes).
131#[derive(Debug, Clone)]
132pub struct FileHandle(pub String);
134pub fn mode_attrs(mode: u32) -> FileAttributes {
135 let mut attrs = FileAttributes::empty();
136 attrs.permissions = Some(mode);
137 attrs
140impl Sftp {
141 /// `openSftp`: new SFTP channel on the connected session.
142 pub async fn open(manager: &SessionManager, session_id: &str) -> CoreResult<Arc<Sftp>> {
143 let record = manager.require_connected(session_id)?;
144 let platform = crate::core::ssh::platform::detect_platform(manager, &record, false).await?;
145 crate::core::matrix::require("files.channel.open", crate::core::matrix::os_of(&platform))?;
146 Self::open_record(&record).await
147 }
149 pub async fn open_record(record: &Arc<SessionRecord>) -> CoreResult<Arc<Sftp>> {
150 let handle = record.handle()?;
151 let failed = |e: String| if record.is_connected() { CoreError::new("SFTP_OPEN_FAILED", e) } else { not_connected() };
152 let opened = tokio::time::timeout(Duration::from_secs(30), async {
153 let channel = handle.channel_open_session().await.map_err(|e| e.to_string())?;
154 channel.request_subsystem(true, "sftp").await.map_err(|e| e.to_string())?;
155 let raw = RawSftpSession::new(channel.into_stream());
156 raw.set_timeout(REQUEST_TIMEOUT_SECS);
157 let version = raw.init().await.map_err(|e| e.to_string())?;
158 // Large requests only when the server announces its limits; Win32-OpenSSH 8.1
159 // (Windows Server 2022) fails 255 KiB writes. Small pieces are sent concurrently.
160 let base = 34_000 - 1024;
161 let (mut max_read, mut max_write) = (base, base);
162 if version.extensions.contains_key("limits@openssh.com") {
163 if let Ok(limits) = raw.limits().await {
164 let large = 256 * 1024 - 1024;
165 max_read = large;
166 max_write = large;
167 if limits.max_read_len > 0 {
168 max_read = max_read.min(limits.max_read_len as usize);
169 }
170 if limits.max_write_len > 0 {
171 max_write = max_write.min(limits.max_write_len as usize);
172 }
173 }
174 }
175 Ok::<_, String>((raw, max_read, max_write))
176 })
177 .await
178 .map_err(|_| failed("SFTP subsystem did not answer.".into()))?
179 .map_err(failed)?;
180 let (raw, max_read, max_write) = opened;
181 Ok(Arc::new(Sftp { raw, max_read, max_write }))
182 }
184 pub fn close(&self) {
185 let _ = self.raw.close_session();
186 }
188 pub async fn stat(&self, path: &str) -> SftpResult<Stat> {
189 self.raw.stat(path).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
190 }
192 pub async fn lstat(&self, path: &str) -> SftpResult<Stat> {
193 self.raw.lstat(path).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
194 }
196 pub async fn exists(&self, path: &str) -> bool {
197 self.lstat(path).await.is_ok()
198 }
200 pub async fn read_dir(&self, path: &str) -> SftpResult<Vec<DirEntry>> {
201 let handle = self.raw.opendir(path).await.map_err(convert)?.handle;
202 let mut entries = Vec::new();
203 let result = loop {
204 match self.raw.readdir(handle.as_str()).await {
205 Ok(name) => {
206 entries.extend(name.files.into_iter().map(|f| DirEntry { stat: Stat::from(&f.attrs), name: f.filename }));
207 if entries.len() > 1_000_000 {
208 break Err(SftpError::Other("Directory listing is too large.".into()));
209 }
210 }
211 Err(RawError::Status(status)) if status.status_code == StatusCode::Eof => break Ok(()),
212 Err(error) => break Err(convert(error)),
213 }
214 };
215 let _ = self.raw.close(handle).await;
216 result.map(|_| entries)
217 }
219 pub async fn mkdir(&self, path: &str, mode: u32) -> SftpResult<()> {
220 self.raw.mkdir(path, mode_attrs(mode)).await.map(|_| ()).map_err(convert)
221 }
223 pub async fn rmdir(&self, path: &str) -> SftpResult<()> {
224 self.raw.rmdir(path).await.map(|_| ()).map_err(convert)
225 }
227 pub async fn unlink(&self, path: &str) -> SftpResult<()> {
228 self.raw.remove(path).await.map(|_| ()).map_err(convert)
229 }
231 pub async fn rename(&self, from: &str, to: &str) -> SftpResult<()> {
232 self.raw.rename(from, to).await.map(|_| ()).map_err(convert)
233 }
235 /// ssh2 flag strings used by 0.1.0: "r", "r+", "wx".
236 pub async fn open_file(&self, path: &str, flags: &str, mode: Option<u32>) -> SftpResult<FileHandle> {
237 let flags = match flags {
238 "r" => OpenFlags::READ,
239 "r+" => OpenFlags::READ | OpenFlags::WRITE,
240 "wx" => OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::EXCLUDE | OpenFlags::TRUNCATE,
241 _ => OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
242 };
243 let attrs = mode.map(mode_attrs).unwrap_or_else(FileAttributes::empty);
244 self.raw.open(path, flags, attrs).await.map(|h| FileHandle(h.handle)).map_err(convert)
245 }
247 pub async fn close_file(&self, handle: &FileHandle) -> SftpResult<()> {
248 self.raw.close(handle.0.as_str()).await.map(|_| ()).map_err(convert)
249 }
251 pub async fn fstat(&self, handle: &FileHandle) -> SftpResult<Stat> {
252 self.raw.fstat(handle.0.as_str()).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
253 }
255 pub async fn truncate(&self, handle: &FileHandle, size: u64) -> SftpResult<()> {
256 let mut attrs = FileAttributes::empty();
257 attrs.size = Some(size);
258 self.raw.fsetstat(handle.0.as_str(), attrs).await.map(|_| ()).map_err(convert)
259 }
261 pub async fn fsync(&self, handle: &FileHandle) -> SftpResult<()> {
262 self.raw.fsync(handle.0.as_str()).await.map(|_| ()).map_err(convert)
263 }
265 /// Reads up to `len` bytes at `offset`; an empty result means end of file.
266 pub async fn read_at(&self, handle: &FileHandle, offset: u64, len: u32) -> SftpResult<Vec<u8>> {
267 match self.raw.read(handle.0.as_str(), offset, len).await {
268 Ok(data) => Ok(data.data),
269 Err(RawError::Status(status)) if status.status_code == StatusCode::Eof => Ok(Vec::new()),
270 Err(error) => Err(convert(error)),
271 }
272 }
274 /// Fills `len` bytes from `offset` (servers may answer short reads), stops at EOF.
275 pub async fn read_full(&self, handle: &FileHandle, offset: u64, len: usize) -> SftpResult<Vec<u8>> {
276 // All pieces in parallel; a short piece (EOF or a server limit) falls back to sequential reads.
277 let pieces: Vec<(u64, usize)> = (0..len).step_by(self.max_read.max(1)).map(|start| (offset + start as u64, (len - start).min(self.max_read))).collect();
278 let results = futures::future::join_all(pieces.iter().map(|(at, want)| self.read_at(handle, *at, *want as u32))).await;
279 let mut out = Vec::with_capacity(len);
280 for ((_, want), result) in pieces.iter().zip(results) {
281 let data = result?;
282 let full = data.len() == *want;
283 out.extend_from_slice(&data);
284 if !full {
285 break;
286 }
287 }
288 while out.len() < len {
289 let want = (len - out.len()).min(self.max_read) as u32;
290 let chunk = self.read_at(handle, offset + out.len() as u64, want).await?;
291 if chunk.is_empty() {
292 break;
293 }
294 out.extend_from_slice(&chunk);
295 }
296 Ok(out)
297 }
299 /// Writes `data` at `offset` in protocol-sized pieces.
300 pub async fn write_at(&self, handle: &FileHandle, offset: u64, data: &[u8]) -> SftpResult<()> {
301 let writes = data.chunks(self.max_write.max(1)).enumerate().map(|(i, piece)| self.raw.write(handle.0.as_str(), offset + (i * self.max_write) as u64, piece.to_vec()));
302 for result in futures::future::join_all(writes).await {
303 result.map_err(convert)?;
304 }
305 Ok(())
306 }
308 /// Whole-file helpers for small files (editor temp copies, probes).
309 pub async fn write_new_file(&self, path: &str, data: &[u8], mode: u32) -> SftpResult<()> {
310 let handle = self.open_file(path, "wx", Some(mode)).await?;
311 let result = self.write_at(&handle, 0, data).await;
312 let closed = self.close_file(&handle).await;
313 result.and(closed)
314 }
317impl Drop for Sftp {
318 fn drop(&mut self) {
319 let _ = self.raw.close_session();
320 }
SHA-2564854ffc71870e4f5a68fbfff2faeaaded34eb6732288788c241744f81d586ba4

CH-J Proprietary Software License 1.14 · Izvor je osiguran pod CH-J Proprietary Software License 1.14. Njena dostupnost ne mijenja uvjete licenciranja niti daje dodatna prava.

Arhiv SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
Datoteke