opendal_service_ftp/
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::fmt::Debug;
19use std::str::FromStr;
20use std::sync::Arc;
21
22use http::Uri;
23use log::debug;
24use suppaftp::FtpError;
25use suppaftp::Status;
26use suppaftp::list::File;
27use suppaftp::types::Response;
28
29use super::FTP_SCHEME;
30use super::config::FtpConfig;
31use super::core::FtpCore;
32use super::core::Manager;
33use super::core::format_ftp_error;
34use super::deleter::FtpDeleter;
35use super::reader::*;
36use opendal_core::raw::*;
37use opendal_core::*;
38
39/// FTP and FTPS services support.
40#[doc = include_str!("docs.md")]
41#[derive(Debug, Default)]
42pub struct FtpBuilder {
43    pub(super) config: FtpConfig,
44}
45
46impl FtpBuilder {
47    /// set endpoint for ftp backend.
48    pub fn endpoint(mut self, endpoint: &str) -> Self {
49        self.config.endpoint = if endpoint.is_empty() {
50            None
51        } else {
52            Some(endpoint.to_string())
53        };
54
55        self
56    }
57
58    /// set root path for ftp backend.
59    pub fn root(mut self, root: &str) -> Self {
60        self.config.root = if root.is_empty() {
61            None
62        } else {
63            Some(root.to_string())
64        };
65
66        self
67    }
68
69    /// set user for ftp backend.
70    pub fn user(mut self, user: &str) -> Self {
71        self.config.user = if user.is_empty() {
72            None
73        } else {
74            Some(user.to_string())
75        };
76
77        self
78    }
79
80    /// set password for ftp backend.
81    pub fn password(mut self, password: &str) -> Self {
82        self.config.password = if password.is_empty() {
83            None
84        } else {
85            Some(password.to_string())
86        };
87
88        self
89    }
90}
91
92impl Builder for FtpBuilder {
93    type Config = FtpConfig;
94
95    fn build(self) -> Result<impl Service> {
96        debug!("ftp backend build started: {:?}", self);
97        let endpoint = match &self.config.endpoint {
98            None => return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")),
99            Some(v) => v,
100        };
101
102        let endpoint_uri = match endpoint.parse::<Uri>() {
103            Err(e) => {
104                return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
105                    .with_context("endpoint", endpoint)
106                    .set_source(e));
107            }
108            Ok(uri) => uri,
109        };
110
111        let host = endpoint_uri.host().unwrap_or("127.0.0.1");
112        let port = endpoint_uri.port_u16().unwrap_or(21);
113
114        let endpoint = format!("{host}:{port}");
115
116        let enable_secure = match endpoint_uri.scheme_str() {
117            Some("ftp") => false,
118            // if the user forgot to add a scheme prefix
119            // treat it as using secured scheme
120            Some("ftps") | None => true,
121
122            Some(s) => {
123                return Err(Error::new(
124                    ErrorKind::ConfigInvalid,
125                    "endpoint is unsupported or invalid",
126                )
127                .with_context("endpoint", s));
128            }
129        };
130
131        let root = normalize_root(&self.config.root.unwrap_or_default());
132
133        let user = match &self.config.user {
134            None => "".to_string(),
135            Some(v) => v.clone(),
136        };
137
138        let password = match &self.config.password {
139            None => "".to_string(),
140            Some(v) => v.clone(),
141        };
142
143        let info = ServiceInfo::new(FTP_SCHEME, &root, "");
144        let capability = Capability {
145            stat: true,
146
147            read: true,
148
149            write: true,
150            write_can_multi: true,
151            write_can_append: true,
152
153            delete: true,
154            create_dir: true,
155
156            list: true,
157
158            shared: true,
159
160            ..Default::default()
161        };
162
163        let manager = Manager {
164            endpoint: endpoint.clone(),
165            root: root.clone(),
166            user: user.clone(),
167            password: password.clone(),
168            enable_secure,
169        };
170
171        let core = Arc::new(FtpCore::new(info, capability, manager.clone()));
172        Ok(FtpBackend { core })
173    }
174}
175
176#[derive(Clone)]
177pub struct FtpBackend {
178    pub(crate) core: Arc<FtpCore>,
179}
180
181impl Debug for FtpBackend {
182    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
183        f.debug_struct("FtpBackend").finish()
184    }
185}
186
187impl Service for FtpBackend {
188    type Reader = oio::StreamReader<FtpReader>;
189    type Writer = FtpLazyWriter;
190    type Lister = FtpLazyLister;
191    type Deleter = oio::OneShotDeleter<FtpDeleter>;
192    type Copier = ();
193
194    fn info(&self) -> ServiceInfo {
195        self.core.info()
196    }
197
198    fn capability(&self) -> Capability {
199        self.core.capability()
200    }
201
202    async fn create_dir(
203        &self,
204        _ctx: &OperationContext,
205        path: &str,
206        _: OpCreateDir,
207    ) -> Result<RpCreateDir> {
208        let mut ftp_stream = self.core.ftp_connect(Operation::CreateDir).await?;
209
210        let paths: Vec<&str> = path.split_inclusive('/').collect();
211
212        let mut curr_path = String::new();
213
214        for path in paths {
215            curr_path.push_str(path);
216            match ftp_stream.mkdir(&curr_path).await {
217                // Do nothing if status is FileUnavailable or OK(()) is return.
218                Err(FtpError::UnexpectedResponse(Response {
219                    status: Status::FileUnavailable,
220                    ..
221                }))
222                | Ok(()) => (),
223                Err(e) => {
224                    return Err(format_ftp_error(e));
225                }
226            }
227        }
228
229        Ok(RpCreateDir::default())
230    }
231
232    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
233        let file = self.ftp_stat(path).await?;
234
235        let mode = if file.is_file() {
236            EntryMode::FILE
237        } else if file.is_directory() {
238            EntryMode::DIR
239        } else {
240            EntryMode::Unknown
241        };
242
243        let mut meta = Metadata::new(mode);
244        meta.set_content_length(file.size() as u64);
245        meta.set_last_modified(Timestamp::try_from(file.modified())?);
246
247        Ok(RpStat::new(meta))
248    }
249    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
250        let output: oio::StreamReader<FtpReader> = {
251            Ok(oio::StreamReader::new(FtpReader::new(
252                self.clone(),
253                path,
254                args,
255            )))
256        }?;
257
258        Ok(output)
259    }
260
261    fn write(&self, _ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
262        Ok(FtpLazyWriter::new(self.core.clone(), path, op))
263    }
264
265    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
266        let output: oio::OneShotDeleter<FtpDeleter> =
267            { Ok(oio::OneShotDeleter::new(FtpDeleter::new(self.core.clone()))) }?;
268
269        Ok(output)
270    }
271
272    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
273        Ok(FtpLazyLister::new(self.core.clone(), path))
274    }
275
276    fn copy(
277        &self,
278        _ctx: &OperationContext,
279        _from: &str,
280        _to: &str,
281        _args: OpCopy,
282        _opts: OpCopier,
283    ) -> Result<Self::Copier> {
284        Err(Error::new(
285            ErrorKind::Unsupported,
286            "operation is not supported",
287        ))
288    }
289
290    async fn rename(
291        &self,
292        _ctx: &OperationContext,
293        _from: &str,
294        _to: &str,
295        _args: OpRename,
296    ) -> Result<RpRename> {
297        Err(Error::new(
298            ErrorKind::Unsupported,
299            "operation is not supported",
300        ))
301    }
302
303    async fn presign(
304        &self,
305        _ctx: &OperationContext,
306        _path: &str,
307        _args: OpPresign,
308    ) -> Result<RpPresign> {
309        Err(Error::new(
310            ErrorKind::Unsupported,
311            "operation is not supported",
312        ))
313    }
314}
315
316impl FtpBackend {
317    pub async fn ftp_stat(&self, path: &str) -> Result<File> {
318        let mut ftp_stream = self.core.ftp_connect(Operation::Stat).await?;
319
320        let (parent, basename) = (get_parent(path), get_basename(path));
321
322        let pathname = if parent == "/" { None } else { Some(parent) };
323
324        let resp = ftp_stream.list(pathname).await.map_err(format_ftp_error)?;
325
326        // Get stat of file.
327        let mut files = resp
328            .into_iter()
329            .filter_map(|file| File::from_str(file.as_str()).ok())
330            .filter(|f| f.name() == basename.trim_end_matches('/'))
331            .collect::<Vec<File>>();
332
333        if files.is_empty() {
334            Err(Error::new(
335                ErrorKind::NotFound,
336                "file is not found during list",
337            ))
338        } else {
339            Ok(files.remove(0))
340        }
341    }
342}
343
344#[cfg(test)]
345mod build_test {
346    use super::FtpBuilder;
347    use crate::FtpConfig;
348    use opendal_core::*;
349
350    #[test]
351    fn test_build() {
352        // ftps scheme, should suffix with default port 21
353        let b = FtpBuilder::default()
354            .endpoint("ftps://ftp_server.local")
355            .build();
356        assert!(b.is_ok());
357
358        // ftp scheme
359        let b = FtpBuilder::default()
360            .endpoint("ftp://ftp_server.local:1234")
361            .build();
362        assert!(b.is_ok());
363
364        // no scheme
365        let b = FtpBuilder::default()
366            .endpoint("ftp_server.local:8765")
367            .build();
368        assert!(b.is_ok());
369
370        // invalid scheme
371        let b = FtpBuilder::default()
372            .endpoint("invalidscheme://ftp_server.local:8765")
373            .build();
374        assert!(b.is_err());
375        let e = b.unwrap_err();
376        assert_eq!(e.kind(), ErrorKind::ConfigInvalid);
377    }
378
379    #[test]
380    fn from_uri_sets_endpoint_and_root() {
381        let uri = OperatorUri::new(
382            "ftp://example.com/public/data",
383            Vec::<(String, String)>::new(),
384        )
385        .unwrap();
386
387        let cfg = FtpConfig::from_uri(&uri).unwrap();
388        assert_eq!(cfg.endpoint.as_deref(), Some("ftp://example.com"));
389        assert_eq!(cfg.root.as_deref(), Some("public/data"));
390    }
391
392    #[test]
393    fn from_uri_applies_credentials_from_query() {
394        let uri = OperatorUri::new(
395            "ftp://example.com/data",
396            vec![
397                ("user".to_string(), "alice".to_string()),
398                ("password".to_string(), "secret".to_string()),
399            ],
400        )
401        .unwrap();
402
403        let cfg = FtpConfig::from_uri(&uri).unwrap();
404        assert_eq!(cfg.endpoint.as_deref(), Some("ftp://example.com"));
405        assert_eq!(cfg.user.as_deref(), Some("alice"));
406        assert_eq!(cfg.password.as_deref(), Some("secret"));
407    }
408}
opendal_service_ftp/backend.rs

opendal_service_ftp/
backend.rs
