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
188 read: true,
189
190 write: true,
191 write_can_multi: true,
192
193 create_dir: true,
194 delete: true,
195
196 list: true,
197 list_with_limit: true,
198
199 copy: self.config.enable_copy,
200 rename: true,
201
202 shared: true,
203
204 ..Default::default()
205 });
206
207 let accessor_info = Arc::new(info);
208 let core = Arc::new(SftpCore {
209 info: accessor_info,
210 endpoint,
211 root,
212 user,
213 key: self.config.key.clone(),
214 known_hosts_strategy,
215
216 client: OnceCell::new(),
217 });
218
219 debug!("sftp backend finished: {:?}", &self);
220 Ok(SftpBackend { core })
221 }
222}
223
224#[derive(Clone)]
226pub struct SftpBackend {
227 pub core: Arc<SftpCore>,
228}
229
230impl Debug for SftpBackend {
231 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
232 f.debug_struct("SftpBackend")
233 .field("core", &self.core)
234 .finish()
235 }
236}
237
238impl Access for SftpBackend {
239 type Reader = SftpReader;
240 type Writer = SftpWriter;
241 type Lister = Option<SftpLister>;
242 type Deleter = oio::OneShotDeleter<SftpDeleter>;
243
244 fn info(&self) -> Arc<AccessorInfo> {
245 self.core.info.clone()
246 }
247
248 async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
249 let client = self.core.connect().await?;
250 let mut fs = client.fs();
251 fs.set_cwd(&self.core.root);
252
253 let paths = Path::new(&path).components();
254 let mut current = PathBuf::from(&self.core.root);
255 for p in paths {
256 current = current.join(p);
257 let res = fs.create_dir(p).await;
258
259 if let Err(e) = res {
260 if !is_sftp_protocol_error(&e) {
262 return Err(parse_sftp_error(e));
263 }
264 }
265 fs.set_cwd(¤t);
266 }
267
268 Ok(RpCreateDir::default())
269 }
270
271 async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
272 let client = self.core.connect().await?;
273 let mut fs = client.fs();
274 fs.set_cwd(&self.core.root);
275
276 let meta: Metadata = fs.metadata(path).await.map_err(parse_sftp_error)?.into();
277
278 Ok(RpStat::new(meta))
279 }
280
281 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
282 let client = self.core.connect().await?;
283
284 let mut fs = client.fs();
285 fs.set_cwd(&self.core.root);
286
287 let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
288
289 let mut f = client
290 .open(path.as_path())
291 .await
292 .map_err(parse_sftp_error)?;
293
294 if args.range().offset() != 0 {
295 f.seek(SeekFrom::Start(args.range().offset()))
296 .await
297 .map_err(new_std_io_error)?;
298 }
299
300 Ok((
301 RpRead::default(),
302 SftpReader::new(client, f, args.range().size()),
303 ))
304 }
305
306 async fn write(&self, path: &str, op: OpWrite) -> Result<(RpWrite, Self::Writer)> {
307 if let Some((dir, _)) = path.rsplit_once('/') {
308 self.create_dir(dir, OpCreateDir::default()).await?;
309 }
310
311 let client = self.core.connect().await?;
312
313 let mut fs = client.fs();
314 fs.set_cwd(&self.core.root);
315 let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
316
317 let mut option = client.options();
318 option.create(true);
319 if op.append() {
320 option.append(true);
321 } else {
322 option.write(true).truncate(true);
323 }
324
325 let file = option.open(path).await.map_err(parse_sftp_error)?;
326
327 Ok((RpWrite::new(), SftpWriter::new(file)))
328 }
329
330 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
331 Ok((
332 RpDelete::default(),
333 oio::OneShotDeleter::new(SftpDeleter::new(self.core.clone())),
334 ))
335 }
336
337 async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
338 let client = self.core.connect().await?;
339 let mut fs = client.fs();
340 fs.set_cwd(&self.core.root);
341
342 let file_path = format!("./{path}");
343
344 let dir = match fs.open_dir(&file_path).await {
345 Ok(dir) => dir,
346 Err(e) => {
347 if is_not_found(&e) {
348 return Ok((RpList::default(), None));
349 } else {
350 return Err(parse_sftp_error(e));
351 }
352 }
353 }
354 .read_dir();
355
356 Ok((
357 RpList::default(),
358 Some(SftpLister::new(dir, path.to_owned())),
359 ))
360 }
361
362 async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
363 let client = self.core.connect().await?;
364
365 let mut fs = client.fs();
366 fs.set_cwd(&self.core.root);
367
368 if let Some((dir, _)) = to.rsplit_once('/') {
369 self.create_dir(dir, OpCreateDir::default()).await?;
370 }
371
372 let src = fs.canonicalize(from).await.map_err(parse_sftp_error)?;
373 let dst = fs.canonicalize(to).await.map_err(parse_sftp_error)?;
374 let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?;
375 let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?;
376
377 src_file
378 .copy_all_to(&mut dst_file)
379 .await
380 .map_err(parse_sftp_error)?;
381
382 Ok(RpCopy::default())
383 }
384
385 async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
386 let client = self.core.connect().await?;
387
388 let mut fs = client.fs();
389 fs.set_cwd(&self.core.root);
390
391 if let Some((dir, _)) = to.rsplit_once('/') {
392 self.create_dir(dir, OpCreateDir::default()).await?;
393 }
394 fs.rename(from, to).await.map_err(parse_sftp_error)?;
395
396 Ok(RpRename::default())
397 }
398}