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            host: host.to_string(),
166            root: root.clone(),
167            user: user.clone(),
168            password: password.clone(),
169            enable_secure,
170        };
171
172        let core = Arc::new(FtpCore::new(info, capability, manager.clone()));
173        Ok(FtpBackend { core })
174    }
175}
176
177#[derive(Clone)]
178pub struct FtpBackend {
179    pub(crate) core: Arc<FtpCore>,
180}
181
182impl Debug for FtpBackend {
183    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
184        f.debug_struct("FtpBackend").finish()
185    }
186}
187
188impl Service for FtpBackend {
189    type Reader = oio::StreamReader<FtpReader>;
190    type Writer = FtpLazyWriter;
191    type Lister = FtpLazyLister;
192    type Deleter = oio::OneShotDeleter<FtpDeleter>;
193    type Copier = ();
194    type Composer = ();
195
196    fn info(&self) -> ServiceInfo {
197        self.core.info()
198    }
199
200    fn capability(&self) -> Capability {
201        self.core.capability()
202    }
203
204    async fn create_dir(
205        &self,
206        _ctx: &OperationContext,
207        path: &str,
208        _: OpCreateDir,
209    ) -> Result<RpCreateDir> {
210        let mut ftp_stream = self.core.ftp_connect(Operation::CreateDir).await?;
211
212        let paths: Vec<&str> = path.split_inclusive('/').collect();
213
214        let mut curr_path = String::new();
215
216        for path in paths {
217            curr_path.push_str(path);
218            match ftp_stream.mkdir(&curr_path).await {
219                // Do nothing if status is FileUnavailable or OK(()) is return.
220                Err(FtpError::UnexpectedResponse(Response {
221                    status: Status::FileUnavailable,
222                    ..
223                }))
224                | Ok(()) => (),
225                Err(e) => {
226                    return Err(format_ftp_error(e));
227                }
228            }
229        }
230
231        Ok(RpCreateDir::default())
232    }
233
234    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
235        let file = self.ftp_stat(path).await?;
236
237        let mode = if file.is_file() {
238            EntryMode::FILE
239        } else if file.is_directory() {
240            EntryMode::DIR
241        } else {
242            EntryMode::Unknown
243        };
244
245        let mut meta = match mode {
246            EntryMode::FILE => MetadataBuilder::file(file.size() as u64),
247            EntryMode::DIR => MetadataBuilder::dir(),
248            EntryMode::Unknown => MetadataBuilder::unknown(),
249        };
250        meta.last_modified(Timestamp::try_from(file.modified())?);
251
252        Ok(RpStat::new(meta.build()))
253    }
254    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
255        let output: oio::StreamReader<FtpReader> = {
256            Ok(oio::StreamReader::new(FtpReader::new(
257                self.clone(),
258                path,
259                args,
260            )))
261        }?;
262
263        Ok(output)
264    }
265
266    fn write(&self, _ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
267        Ok(FtpLazyWriter::new(self.core.clone(), path, op))
268    }
269
270    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
271        let output: oio::OneShotDeleter<FtpDeleter> =
272            { Ok(oio::OneShotDeleter::new(FtpDeleter::new(self.core.clone()))) }?;
273
274        Ok(output)
275    }
276
277    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
278        Ok(FtpLazyLister::new(self.core.clone(), path))
279    }
280
281    fn copy(
282        &self,
283        _ctx: &OperationContext,
284        _from: &str,
285        _to: &str,
286        _args: OpCopy,
287    ) -> Result<Self::Copier> {
288        Err(Error::new(
289            ErrorKind::Unsupported,
290            "operation is not supported",
291        ))
292    }
293
294    async fn rename(
295        &self,
296        _ctx: &OperationContext,
297        _from: &str,
298        _to: &str,
299        _args: OpRename,
300    ) -> Result<RpRename> {
301        Err(Error::new(
302            ErrorKind::Unsupported,
303            "operation is not supported",
304        ))
305    }
306
307    async fn presign(
308        &self,
309        _ctx: &OperationContext,
310        _path: &str,
311        _args: OpPresign,
312    ) -> Result<RpPresign> {
313        Err(Error::new(
314            ErrorKind::Unsupported,
315            "operation is not supported",
316        ))
317    }
318}
319
320impl FtpBackend {
321    pub async fn ftp_stat(&self, path: &str) -> Result<File> {
322        let mut ftp_stream = self.core.ftp_connect(Operation::Stat).await?;
323
324        let (parent, basename) = (get_parent(path), get_basename(path));
325
326        let pathname = if parent == "/" { None } else { Some(parent) };
327
328        let resp = ftp_stream.list(pathname).await.map_err(format_ftp_error)?;
329
330        // Get stat of file.
331        let mut files = resp
332            .into_iter()
333            .filter_map(|file| File::from_str(file.as_str()).ok())
334            .filter(|f| f.name() == basename.trim_end_matches('/'))
335            .collect::<Vec<File>>();
336
337        if files.is_empty() {
338            Err(Error::new(
339                ErrorKind::NotFound,
340                "file is not found during list",
341            ))
342        } else {
343            Ok(files.remove(0))
344        }
345    }
346}
347
348#[cfg(test)]
349mod build_test {
350    use super::FtpBuilder;
351    use crate::FtpConfig;
352    use opendal_core::*;
353
354    #[test]
355    fn test_build() {
356        // ftps scheme, should suffix with default port 21
357        let b = FtpBuilder::default()
358            .endpoint("ftps://ftp_server.local")
359            .build();
360        assert!(b.is_ok());
361
362        // ftp scheme
363        let b = FtpBuilder::default()
364            .endpoint("ftp://ftp_server.local:1234")
365            .build();
366        assert!(b.is_ok());
367
368        // no scheme
369        let b = FtpBuilder::default()
370            .endpoint("ftp_server.local:8765")
371            .build();
372        assert!(b.is_ok());
373
374        // invalid scheme
375        let b = FtpBuilder::default()
376            .endpoint("invalidscheme://ftp_server.local:8765")
377            .build();
378        assert!(b.is_err());
379        let e = b.unwrap_err();
380        assert_eq!(e.kind(), ErrorKind::ConfigInvalid);
381    }
382
383    #[test]
384    fn from_uri_sets_endpoint_and_root() {
385        let uri = OperatorUri::new(
386            "ftp://example.com/public/data",
387            Vec::<(String, String)>::new(),
388        )
389        .unwrap();
390
391        let cfg = FtpConfig::from_uri(&uri).unwrap();
392        assert_eq!(cfg.endpoint.as_deref(), Some("ftp://example.com"));
393        assert_eq!(cfg.root.as_deref(), Some("public/data"));
394    }
395
396    #[test]
397    fn from_uri_applies_credentials_from_query() {
398        let uri = OperatorUri::new(
399            "ftp://example.com/data",
400            vec![
401                ("user".to_string(), "alice".to_string()),
402                ("password".to_string(), "secret".to_string()),
403            ],
404        )
405        .unwrap();
406
407        let cfg = FtpConfig::from_uri(&uri).unwrap();
408        assert_eq!(cfg.endpoint.as_deref(), Some("ftp://example.com"));
409        assert_eq!(cfg.user.as_deref(), Some("alice"));
410        assert_eq!(cfg.password.as_deref(), Some("secret"));
411    }
412}
opendal_service_ftp/backend.rs

opendal_service_ftp/
backend.rs
