opendal/services/sftp/
backend.rs1use std::fmt::Debug;
19use std::fmt::Formatter;
20use std::io::SeekFrom;
21use std::path::Path;
22use std::path::PathBuf;
23use std::sync::Arc;
24
25use log::debug;
26use openssh::KnownHosts;
27use tokio::io::AsyncSeekExt;
28use tokio::sync::OnceCell;
29
30use super::core::SftpCore;
31use super::delete::SftpDeleter;
32use super::error::is_not_found;
33use super::error::is_sftp_protocol_error;
34use super::error::parse_sftp_error;
35use super::lister::SftpLister;
36use super::reader::SftpReader;
37use super::writer::SftpWriter;
38use crate::raw::*;
39use crate::services::SftpConfig;
40use crate::*;
41
42impl Configurator for SftpConfig {
43 type Builder = SftpBuilder;
44 fn into_builder(self) -> Self::Builder {
45 SftpBuilder { config: self }
46 }
47}
48
49#[doc = include_str!("docs.md")]
59#[derive(Default)]
60pub struct SftpBuilder {
61 config: SftpConfig,
62}
63
64impl Debug for SftpBuilder {
65 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
66 f.debug_struct("SftpBuilder")
67 .field("config", &self.config)
68 .finish()
69 }
70}
71
72impl SftpBuilder {
73 pub fn endpoint(mut self, endpoint: &str) -> Self {
76 self.config.endpoint = if endpoint.is_empty() {
77 None
78 } else {
79 Some(endpoint.to_string())
80 };
81
82 self
83 }
84
85 pub fn root(mut self, root: &str) -> Self {
88 self.config.root = if root.is_empty() {
89 None
90 } else {
91 Some(root.to_string())
92 };
93
94 self
95 }
96
97 pub fn user(mut self, user: &str) -> Self {
99 self.config.user = if user.is_empty() {
100 None
101 } else {
102 Some(user.to_string())
103 };
104
105 self
106 }
107
108 pub fn key(mut self, key: &str) -> Self {
110 self.config.key = if key.is_empty() {
111 None
112 } else {
113 Some(key.to_string())
114 };
115
116 self
117 }
118
119 pub fn known_hosts_strategy(mut self, strategy: &str) -> Self {
125 self.config.known_hosts_strategy = if strategy.is_empty() {
126 None
127 } else {
128 Some(strategy.to_string())
129 };
130
131 self
132 }
133
134 pub fn enable_copy(mut self, enable_copy: bool) -> Self {
137 self.config.enable_copy = enable_copy;
138
139 self
140 }
141}
142
143impl Builder for SftpBuilder {
144 const SCHEME: Scheme = Scheme::Sftp;
145 type Config = SftpConfig;
146
147 fn build(self) -> Result<impl Access> {
148 debug!("sftp backend build started: {:?}", &self);
149 let endpoint = match self.config.endpoint.clone() {
150 Some(v) => v,
151 None => return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")),
152 };
153
154 let user = self.config.user.clone();
155
156 let root = self
157 .config
158 .root
159 .clone()
160 .map(|r| normalize_root(r.as_str()))
161 .unwrap_or_default();
162
163 let known_hosts_strategy = match &self.config.known_hosts_strategy {
164 Some(v) => {
165 let v = v.to_lowercase();
166 if v == "strict" {
167 KnownHosts::Strict
168 } else if v == "accept" {
169 KnownHosts::Accept
170 } else if v == "add" {
171 KnownHosts::Add
172 } else {
173 return Err(Error::new(
174 ErrorKind::ConfigInvalid,
175 format!("unknown known_hosts strategy: {}", v).as_str(),
176 ));
177 }
178 }
179 None => KnownHosts::Strict,
180 };
181
182 let info = AccessorInfo::default();
183 info.set_root(root.as_str())
184 .set_scheme(Scheme::Sftp)
185 .set_native_capability(Capability {
186 stat: true,
187 stat_has_content_length: true,
188 stat_has_last_modified: true,
189
190 read: true,
191
192 write: true,
193 write_can_multi: true,
194
195 create_dir: true,
196 delete: true,
197
198 list: true,
199 list_with_limit: true,
200 list_has_content_length: true,
201 list_has_last_modified: true,
202
203 copy: self.config.enable_copy,
204 rename: true,
205
206 shared: true,
207
208 ..Default::default()
209 });
210
211 let accessor_info = Arc::new(info);
212 let core = Arc::new(SftpCore {
213 info: accessor_info,
214 endpoint,
215 root,
216 user,
217 key: self.config.key.clone(),
218 known_hosts_strategy,
219
220 client: OnceCell::new(),
221 });
222
223 debug!("sftp backend finished: {:?}", &self);
224 Ok(SftpBackend { core })
225 }
226}
227
228#[derive(Clone)]
230pub struct SftpBackend {
231 pub core: Arc<SftpCore>,
232}
233
234impl Debug for SftpBackend {
235 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
236 f.debug_struct("SftpBackend")
237 .field("core", &self.core)
238 .finish()
239 }
240}
241
242impl Access for SftpBackend {
243 type Reader = SftpReader;
244 type Writer = SftpWriter;
245 type Lister = Option<SftpLister>;
246 type Deleter = oio::OneShotDeleter<SftpDeleter>;
247 type BlockingReader = ();
248 type BlockingWriter = ();
249 type BlockingLister = ();
250 type BlockingDeleter = ();
251
252 fn info(&self) -> Arc<AccessorInfo> {
253 self.core.info.clone()
254 }
255
256 async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
257 let client = self.core.connect().await?;
258 let mut fs = client.fs();
259 fs.set_cwd(&self.core.root);
260
261 let paths = Path::new(&path).components();
262 let mut current = PathBuf::from(&self.core.root);
263 for p in paths {
264 current = current.join(p);
265 let res = fs.create_dir(p).await;
266
267 if let Err(e) = res {
268 if !is_sftp_protocol_error(&e) {
270 return Err(parse_sftp_error(e));
271 }
272 }
273 fs.set_cwd(¤t);
274 }
275
276 Ok(RpCreateDir::default())
277 }
278
279 async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
280 let client = self.core.connect().await?;
281 let mut fs = client.fs();
282 fs.set_cwd(&self.core.root);
283
284 let meta: Metadata = fs.metadata(path).await.map_err(parse_sftp_error)?.into();
285
286 Ok(RpStat::new(meta))
287 }
288
289 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
290 let client = self.core.connect().await?;
291
292 let mut fs = client.fs();
293 fs.set_cwd(&self.core.root);
294
295 let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
296
297 let mut f = client
298 .open(path.as_path())
299 .await
300 .map_err(parse_sftp_error)?;
301
302 if args.range().offset() != 0 {
303 f.seek(SeekFrom::Start(args.range().offset()))
304 .await
305 .map_err(new_std_io_error)?;
306 }
307
308 Ok((
309 RpRead::default(),
310 SftpReader::new(client, f, args.range().size()),
311 ))
312 }
313
314 async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
315 if let Some((dir, _)) = path.rsplit_once('/') {
316 self.create_dir(dir, OpCreateDir::default()).await?;
317 }
318
319 let client = self.core.connect().await?;
320
321 let mut fs = client.fs();
322 fs.set_cwd(&self.core.root);
323 let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
324
325 let mut option = client.options();
326 option.create(true);
327 if op.append() {
328 option.append(true);
329 } else {
330 option.write(true).truncate(true);
331 }
332
333 let file = option.open(path).await.map_err(parse_sftp_error)?;
334
335 Ok((RpWrite::new(), SftpWriter::new(file)))
336 }
337
338 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
339 Ok((
340 RpDelete::default(),
341 oio::OneShotDeleter::new(SftpDeleter::new(self.core.clone())),
342 ))
343 }
344
345 async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
346 let client = self.core.connect().await?;
347 let mut fs = client.fs();
348 fs.set_cwd(&self.core.root);
349
350 let file_path = format!("./{}", path);
351
352 let dir = match fs.open_dir(&file_path).await {
353 Ok(dir) => dir,
354 Err(e) => {
355 if is_not_found(&e) {
356 return Ok((RpList::default(), None));
357 } else {
358 return Err(parse_sftp_error(e));
359 }
360 }
361 }
362 .read_dir();
363
364 Ok((
365 RpList::default(),
366 Some(SftpLister::new(dir, path.to_owned())),
367 ))
368 }
369
370 async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
371 let client = self.core.connect().await?;
372
373 let mut fs = client.fs();
374 fs.set_cwd(&self.core.root);
375
376 if let Some((dir, _)) = to.rsplit_once('/') {
377 self.create_dir(dir, OpCreateDir::default()).await?;
378 }
379
380 let src = fs.canonicalize(from).await.map_err(parse_sftp_error)?;
381 let dst = fs.canonicalize(to).await.map_err(parse_sftp_error)?;
382 let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?;
383 let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?;
384
385 src_file
386 .copy_all_to(&mut dst_file)
387 .await
388 .map_err(parse_sftp_error)?;
389
390 Ok(RpCopy::default())
391 }
392
393 async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
394 let client = self.core.connect().await?;
395
396 let mut fs = client.fs();
397 fs.set_cwd(&self.core.root);
398
399 if let Some((dir, _)) = to.rsplit_once('/') {
400 self.create_dir(dir, OpCreateDir::default()).await?;
401 }
402 fs.rename(from, to).await.map_err(parse_sftp_error)?;
403
404 Ok(RpRename::default())
405 }
406}