opendal_service_sftp/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::path::Path;
19use std::path::PathBuf;
20use std::sync::Arc;
21
22use log::debug;
23use openssh::KnownHosts;
24
25use super::SFTP_SCHEME;
26use super::config::SftpConfig;
27use super::core::SftpCore;
28use super::core::is_sftp_protocol_error;
29use super::core::parse_sftp_error;
30use super::core::to_metadata;
31use super::deleter::SftpDeleter;
32use super::reader::*;
33use opendal_core::raw::*;
34use opendal_core::*;
35
36/// SFTP services support. (only works on unix)
37///
38/// If you are interested in working on windows, please refer to [this](https://github.com/apache/opendal/issues/2963) issue.
39/// Welcome to leave your comments or make contributions.
40///
41/// Warning: Maximum number of file holdings is depending on the remote system configuration.
42///
43/// For example, the default value is 255 in macOS, and 1024 in linux. If you want to open
44/// lots of files, you should pay attention to close the file after using it.
45#[doc = include_str!("docs.md")]
46#[derive(Debug, Default)]
47pub struct SftpBuilder {
48    pub(super) config: SftpConfig,
49}
50
51impl SftpBuilder {
52    /// set endpoint for sftp backend.
53    /// The format is same as `openssh`, using either `[user@]hostname` or `ssh://[user@]hostname[:port]`. A username or port that is specified in the endpoint overrides the one set in the builder (but does not change the builder).
54    pub fn endpoint(mut self, endpoint: &str) -> Self {
55        self.config.endpoint = if endpoint.is_empty() {
56            None
57        } else {
58            Some(endpoint.to_string())
59        };
60
61        self
62    }
63
64    /// set root path for sftp backend.
65    /// It uses the default directory set by the remote `sftp-server` as default.
66    pub fn root(mut self, root: &str) -> Self {
67        self.config.root = if root.is_empty() {
68            None
69        } else {
70            Some(root.to_string())
71        };
72
73        self
74    }
75
76    /// set user for sftp backend.
77    pub fn user(mut self, user: &str) -> Self {
78        self.config.user = if user.is_empty() {
79            None
80        } else {
81            Some(user.to_string())
82        };
83
84        self
85    }
86
87    /// set key path for sftp backend.
88    pub fn key(mut self, key: &str) -> Self {
89        self.config.key = if key.is_empty() {
90            None
91        } else {
92            Some(key.to_string())
93        };
94
95        self
96    }
97
98    /// set known_hosts strategy for sftp backend.
99    /// available values:
100    /// - Strict (default)
101    /// - Accept
102    /// - Add
103    pub fn known_hosts_strategy(mut self, strategy: &str) -> Self {
104        self.config.known_hosts_strategy = if strategy.is_empty() {
105            None
106        } else {
107            Some(strategy.to_string())
108        };
109
110        self
111    }
112
113    /// Deprecated: SFTP copy capability is enabled by default.
114    #[deprecated(
115        since = "0.57.0",
116        note = "SFTP copy capability is enabled by default and this option is no longer needed."
117    )]
118    pub fn enable_copy(self, _enable_copy: bool) -> Self {
119        self
120    }
121}
122
123impl Builder for SftpBuilder {
124    type Config = SftpConfig;
125
126    fn build(self) -> Result<impl Service> {
127        debug!("sftp backend build started: {:?}", self);
128        let endpoint = match self.config.endpoint.clone() {
129            Some(v) => v,
130            None => return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")),
131        };
132
133        let user = self.config.user.clone();
134
135        let root = self
136            .config
137            .root
138            .clone()
139            .map(|r| normalize_root(r.as_str()))
140            .unwrap_or_default();
141
142        let known_hosts_strategy = match &self.config.known_hosts_strategy {
143            Some(v) => {
144                let v = v.to_lowercase();
145                if v == "strict" {
146                    KnownHosts::Strict
147                } else if v == "accept" {
148                    KnownHosts::Accept
149                } else if v == "add" {
150                    KnownHosts::Add
151                } else {
152                    return Err(Error::new(
153                        ErrorKind::ConfigInvalid,
154                        format!("unknown known_hosts strategy: {v}").as_str(),
155                    ));
156                }
157            }
158            None => KnownHosts::Strict,
159        };
160
161        let info = ServiceInfo::new(SFTP_SCHEME, root.as_str(), "");
162        let capability = Capability {
163            stat: true,
164
165            read: true,
166
167            write: true,
168            write_can_multi: true,
169            write_with_if_not_exists: true,
170
171            create_dir: true,
172            delete: true,
173
174            list: true,
175            list_with_limit: true,
176
177            copy: true,
178            rename: true,
179
180            shared: true,
181
182            ..Default::default()
183        };
184
185        let core = Arc::new(SftpCore::new(
186            info,
187            capability,
188            endpoint,
189            root,
190            user,
191            self.config.key.clone(),
192            known_hosts_strategy,
193        ));
194
195        debug!("sftp backend finished: {:?}", self);
196        Ok(SftpBackend { core })
197    }
198}
199
200#[derive(Clone, Debug)]
201pub struct SftpBackend {
202    pub core: Arc<SftpCore>,
203}
204
205impl Service for SftpBackend {
206    type Reader = oio::StreamReader<SftpReader>;
207    type Writer = SftpLazyWriter;
208    type Lister = SftpLazyLister;
209    type Deleter = oio::OneShotDeleter<SftpDeleter>;
210    type Copier = oio::OneShotCopier;
211    type Composer = ();
212
213    fn info(&self) -> ServiceInfo {
214        self.core.info.clone()
215    }
216
217    fn capability(&self) -> Capability {
218        self.core.capability
219    }
220
221    async fn create_dir(
222        &self,
223        _ctx: &OperationContext,
224        path: &str,
225        _: OpCreateDir,
226    ) -> Result<RpCreateDir> {
227        let client = self.core.connect().await?;
228        let mut fs = client.fs();
229        fs.set_cwd(&self.core.root);
230
231        let paths = Path::new(&path).components();
232        let mut current = PathBuf::from(&self.core.root);
233        for p in paths {
234            current = current.join(p);
235            let res = fs.create_dir(p).await;
236
237            if let Err(e) = res {
238                // ignore error if dir already exists
239                if !is_sftp_protocol_error(&e) {
240                    return Err(parse_sftp_error(e));
241                }
242            }
243            fs.set_cwd(&current);
244        }
245
246        Ok(RpCreateDir::default())
247    }
248
249    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
250        let client = self.core.connect().await?;
251        let mut fs = client.fs();
252        fs.set_cwd(&self.core.root);
253
254        let meta: Metadata = to_metadata(fs.metadata(path).await.map_err(parse_sftp_error)?)?;
255
256        Ok(RpStat::new(meta))
257    }
258    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
259        let output: oio::StreamReader<SftpReader> = {
260            Ok(oio::StreamReader::new(SftpReader::new(
261                self.clone(),
262                path,
263                args,
264            )))
265        }?;
266
267        Ok(output)
268    }
269
270    fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
271        Ok(SftpLazyWriter::new(self.clone(), ctx.clone(), path, op))
272    }
273
274    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
275        let output: oio::OneShotDeleter<SftpDeleter> = {
276            Ok(oio::OneShotDeleter::new(SftpDeleter::new(
277                self.core.clone(),
278            )))
279        }?;
280
281        Ok(output)
282    }
283
284    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
285        Ok(SftpLazyLister::new(self.clone(), path))
286    }
287
288    fn copy(
289        &self,
290        ctx: &OperationContext,
291        from: &str,
292        to: &str,
293        args: OpCopy,
294    ) -> Result<Self::Copier> {
295        let backend = self.clone();
296        let ctx = ctx.clone();
297        let from = from.to_string();
298        let to = to.to_string();
299        Ok(oio::OneShotCopier::new(async move {
300            let client = backend.core.connect().await?;
301
302            let mut fs = client.fs();
303            fs.set_cwd(&backend.core.root);
304
305            if let Some((dir, _)) = to.rsplit_once('/') {
306                backend
307                    .create_dir(&ctx, dir, OpCreateDir::default())
308                    .await?;
309            }
310
311            let src = fs.canonicalize(&from).await.map_err(parse_sftp_error)?;
312            let dst = fs.canonicalize(&to).await.map_err(parse_sftp_error)?;
313            let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?;
314            let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?;
315
316            let source_size = match args.source_content_length_hint() {
317                Some(size) => size,
318                None => src_file
319                    .metadata()
320                    .await
321                    .map_err(parse_sftp_error)?
322                    .len()
323                    .ok_or_else(|| {
324                        Error::new(
325                            ErrorKind::Unexpected,
326                            "sftp source metadata does not contain file length",
327                        )
328                    })?,
329            };
330
331            src_file
332                .copy_all_to(&mut dst_file)
333                .await
334                .map_err(parse_sftp_error)?;
335
336            Ok(MetadataBuilder::file(source_size).build())
337        }))
338    }
339
340    async fn rename(
341        &self,
342        ctx: &OperationContext,
343        from: &str,
344        to: &str,
345        _: OpRename,
346    ) -> Result<RpRename> {
347        let client = self.core.connect().await?;
348
349        let mut fs = client.fs();
350        fs.set_cwd(&self.core.root);
351
352        if let Some((dir, _)) = to.rsplit_once('/') {
353            self.create_dir(ctx, dir, OpCreateDir::default()).await?;
354        }
355        fs.rename(from, to).await.map_err(parse_sftp_error)?;
356
357        Ok(RpRename::default())
358    }
359
360    async fn presign(
361        &self,
362        _ctx: &OperationContext,
363        _path: &str,
364        _args: OpPresign,
365    ) -> Result<RpPresign> {
366        Err(Error::new(
367            ErrorKind::Unsupported,
368            "operation is not supported",
369        ))
370    }
371}
opendal_service_sftp/backend.rs

opendal_service_sftp/
backend.rs
