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
212    fn info(&self) -> ServiceInfo {
213        self.core.info.clone()
214    }
215
216    fn capability(&self) -> Capability {
217        self.core.capability
218    }
219
220    async fn create_dir(
221        &self,
222        _ctx: &OperationContext,
223        path: &str,
224        _: OpCreateDir,
225    ) -> Result<RpCreateDir> {
226        let client = self.core.connect().await?;
227        let mut fs = client.fs();
228        fs.set_cwd(&self.core.root);
229
230        let paths = Path::new(&path).components();
231        let mut current = PathBuf::from(&self.core.root);
232        for p in paths {
233            current = current.join(p);
234            let res = fs.create_dir(p).await;
235
236            if let Err(e) = res {
237                // ignore error if dir already exists
238                if !is_sftp_protocol_error(&e) {
239                    return Err(parse_sftp_error(e));
240                }
241            }
242            fs.set_cwd(&current);
243        }
244
245        Ok(RpCreateDir::default())
246    }
247
248    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
249        let client = self.core.connect().await?;
250        let mut fs = client.fs();
251        fs.set_cwd(&self.core.root);
252
253        let meta: Metadata = to_metadata(fs.metadata(path).await.map_err(parse_sftp_error)?);
254
255        Ok(RpStat::new(meta))
256    }
257    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
258        let output: oio::StreamReader<SftpReader> = {
259            Ok(oio::StreamReader::new(SftpReader::new(
260                self.clone(),
261                path,
262                args,
263            )))
264        }?;
265
266        Ok(output)
267    }
268
269    fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
270        Ok(SftpLazyWriter::new(self.clone(), ctx.clone(), path, op))
271    }
272
273    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
274        let output: oio::OneShotDeleter<SftpDeleter> = {
275            Ok(oio::OneShotDeleter::new(SftpDeleter::new(
276                self.core.clone(),
277            )))
278        }?;
279
280        Ok(output)
281    }
282
283    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
284        Ok(SftpLazyLister::new(self.clone(), path))
285    }
286
287    fn copy(
288        &self,
289        ctx: &OperationContext,
290        from: &str,
291        to: &str,
292        _: OpCopy,
293        _opts: OpCopier,
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            src_file
317                .copy_all_to(&mut dst_file)
318                .await
319                .map_err(parse_sftp_error)?;
320
321            Ok(Metadata::default())
322        }))
323    }
324
325    async fn rename(
326        &self,
327        ctx: &OperationContext,
328        from: &str,
329        to: &str,
330        _: OpRename,
331    ) -> Result<RpRename> {
332        let client = self.core.connect().await?;
333
334        let mut fs = client.fs();
335        fs.set_cwd(&self.core.root);
336
337        if let Some((dir, _)) = to.rsplit_once('/') {
338            self.create_dir(ctx, dir, OpCreateDir::default()).await?;
339        }
340        fs.rename(from, to).await.map_err(parse_sftp_error)?;
341
342        Ok(RpRename::default())
343    }
344
345    async fn presign(
346        &self,
347        _ctx: &OperationContext,
348        _path: &str,
349        _args: OpPresign,
350    ) -> Result<RpPresign> {
351        Err(Error::new(
352            ErrorKind::Unsupported,
353            "operation is not supported",
354        ))
355    }
356}
opendal_service_sftp/backend.rs

opendal_service_sftp/
backend.rs
