feat: implement APPE (append) command per RFC 959

README.md

@@ -24,7 +24,7 @@ possible.
 
 Feature highlights:
 
-* 41 Supported FTP commands (see [commands directory](./src/server/controlchan/commands)) and growing
+* 42 Supported FTP commands (see [commands directory](./src/server/controlchan/commands)) and growing
 * Ability to implement own storage back-ends
 * Ability to implement own authentication back-ends
 * Explicit FTPS (TLS)

src/server/chancomms.rs

@@ -31,6 +31,10 @@ pub enum DataChanCmd {
         /// The path to the file the client would like to store.
         path: String,
     },
+    Appe {
+        /// The path to the file the client would like to append to.
+        path: String,
+    },
     List {
         /// Arguments passed along with the list command.
         options: Option<String>,
@@ -53,6 +57,7 @@ impl DataChanCmd {
         match self {
             DataChanCmd::Retr { path, .. } => Some(path.clone()),
             DataChanCmd::Stor { path, .. } => Some(path.clone()),
+            DataChanCmd::Appe { path, .. } => Some(path.clone()),
             DataChanCmd::List { path, .. } => path.clone(),
             DataChanCmd::Mlsd { path } => path.clone(),
             DataChanCmd::Nlst { path } => path.clone(),

src/server/controlchan/command.rs

@@ -52,6 +52,10 @@ pub enum Command {
         /// The path to the file the client would like to store.
         path: String,
     },
+    Appe {
+        /// The path to the file the client would like to append to.
+        path: String,
+    },
     List {
         /// Arguments passed along with the list command.
         options: Option<String>,

src/server/controlchan/commands/appe.rs

@@ -0,0 +1,60 @@
+//! The RFC 959 Append (`APPE`) command
+//
+// This command causes the server-DTP to accept the data
+// transferred via the data connection and to store the data in
+// a file at the server site.  If the file specified in the
+// pathname exists at the server site, the data shall be
+// appended to that file; otherwise the file shall be created.
+
+use crate::server::chancomms::DataChanCmd;
+use crate::{
+    auth::UserDetail,
+    server::controlchan::{
+        Reply, ReplyCode,
+        command::Command,
+        error::ControlChanError,
+        handler::{CommandContext, CommandHandler},
+    },
+    storage::{Metadata, StorageBackend},
+};
+use async_trait::async_trait;
+
+#[derive(Debug)]
+pub struct Appe;
+
+#[async_trait]
+impl<Storage, User> CommandHandler<Storage, User> for Appe
+where
+    User: UserDetail + 'static,
+    Storage: StorageBackend<User> + 'static,
+    Storage::Metadata: Metadata,
+{
+    #[tracing_attributes::instrument]
+    async fn handle(&self, args: CommandContext<Storage, User>) -> Result<Reply, ControlChanError> {
+        let mut session = args.session.lock().await;
+
+        let (cmd, path): (DataChanCmd, String) = match args.parsed_command.clone() {
+            Command::Appe { path } => {
+                let path_clone = path.clone();
+                (DataChanCmd::Appe { path }, path_clone)
+            }
+            _ => panic!("Programmer error, expected command to be APPE"),
+        };
+
+        let logger = args.logger;
+        match session.data_cmd_tx.take() {
+            Some(tx) => {
+                tokio::spawn(async move {
+                    if let Err(err) = tx.send(cmd).await {
+                        slog::warn!(logger, "APPE: could not notify data channel. {}", err);
+                    }
+                });
+                Ok(Reply::new(ReplyCode::FileStatusOkay, "Ready to receive data"))
+            }
+            None => {
+                slog::warn!(logger, "APPE: no data connection established for APPEing {:?}", path);
+                Ok(Reply::new(ReplyCode::CantOpenDataConnection, "No data connection established"))
+            }
+        }
+    }
+}

src/server/controlchan/commands/mod.rs

@@ -7,6 +7,7 @@
 mod abor;
 mod acct;
 mod allo;
+mod appe;
 mod auth;
 mod ccc;
 mod cdup;
@@ -50,6 +51,7 @@ pub use self::md5::Md5;
 pub use abor::Abor;
 pub use acct::Acct;
 pub use allo::Allo;
+pub use appe::Appe;
 pub use auth::{Auth, AuthParam};
 pub use ccc::Ccc;
 pub use cdup::Cdup;

src/server/controlchan/control_loop.rs

@@ -472,6 +472,7 @@ where
             Command::Md5 { file } => Box::new(commands::Md5::new(file)),
             Command::Mlst { path } => Box::new(commands::Mlst::new(path)),
             Command::Mlsd { .. } => Box::new(commands::Mlsd),
+            Command::Appe { .. } => Box::new(commands::Appe),
             Command::Other { .. } => return Ok(Reply::new(ReplyCode::CommandSyntaxError, "Command not implemented")),
         };
 

src/server/controlchan/line_parser/parser.rs

@@ -114,6 +114,14 @@ where
             let path = String::from_utf8_lossy(&path);
             Command::Stor { path: path.to_string() }
         }
+        "APPE" => {
+            let path = parse_to_eol(cmd_params)?;
+            if path.is_empty() {
+                return Err(ParseErrorKind::InvalidCommand.into());
+            }
+            let path = String::from_utf8_lossy(&path);
+            Command::Appe { path: path.to_string() }
+        }
         "LIST" => {
             let line = parse_to_eol(cmd_params)?;
             let path = line

src/server/controlchan/line_parser/tests.rs

@@ -600,3 +600,38 @@ fn parse_site() {
         assert_eq!(parse(test.input), test.expected);
     }
 }
+
+#[test]
+fn parse_appe() {
+    struct Test {
+        input: &'static str,
+        expected: Result<Command>,
+    }
+    let tests = [
+        Test {
+            input: "APPE\r\n",
+            expected: Err(ParseErrorKind::InvalidCommand.into()),
+        },
+        Test {
+            input: "APPE \r\n",
+            expected: Err(ParseErrorKind::InvalidCommand.into()),
+        },
+        Test {
+            input: "APPE file.txt\r\n",
+            expected: Ok(Command::Appe { path: "file.txt".into() }),
+        },
+        Test {
+            input: "APPE path/to/file.txt\r\n",
+            expected: Ok(Command::Appe {
+                path: "path/to/file.txt".into(),
+            }),
+        },
+        Test {
+            input: "appe file.txt\r\n",
+            expected: Ok(Command::Appe { path: "file.txt".into() }),
+        },
+    ];
+    for test in tests.iter() {
+        assert_eq!(parse(test.input), test.expected);
+    }
+}

src/server/datachan.rs

@@ -149,6 +149,9 @@ where
             DataChanCmd::Stor { path } => {
                 self.exec_stor(path, start_pos).await;
             }
+            DataChanCmd::Appe { path } => {
+                self.exec_appe(path).await;
+            }
             DataChanCmd::List { path, .. } => {
                 self.exec_list_variant(path, ListCommand::List).await;
             }
@@ -308,6 +311,60 @@ where
         }
     }
 
+    #[tracing_attributes::instrument]
+    async fn exec_appe(self, path: String) {
+        let path_copy = path.clone();
+        let full_path = self.cwd.join(&path);
+        let tx = self.control_msg_tx.clone();
+
+        // Get current file size, or 0 if file doesn't exist
+        let start_pos = match self.storage.metadata((*self.user).as_ref().unwrap(), &full_path).await {
+            Ok(meta) => meta.len(),
+            Err(_) => 0,
+        };
+
+        let start_time = Instant::now();
+        let put_result = self
+            .storage
+            .put(
+                (*self.user).as_ref().unwrap(),
+                Self::reader(self.socket, self.ftps_mode, "appe").await,
+                full_path,
+                start_pos,
+            )
+            .await;
+        let duration = start_time.elapsed();
+
+        match put_result {
+            Ok(bytes) => {
+                slog::info!(
+                    self.logger,
+                    "Successful APPE {:?}; Duration {}; Bytes copied {}; Transfer speed {}; start_pos={}",
+                    &path_copy,
+                    HumanDuration(duration),
+                    HumanBytes(bytes),
+                    TransferSpeed(bytes as f64 / duration.as_secs_f64()),
+                    start_pos,
+                );
+
+                metrics::inc_transferred("appe", "success");
+
+                if let Err(err) = tx.send(ControlChanMsg::WrittenData { bytes, path: path_copy }).await {
+                    slog::error!(self.logger, "Could not notify control channel of successful APPE: {:?}", err);
+                }
+            }
+            Err(err) => {
+                slog::warn!(self.logger, "Error during APPE transfer after {}: {:?}", HumanDuration(duration), err);
+
+                categorize_and_register_error(&self.logger, &err, "appe");
+
+                if let Err(err) = tx.send(ControlChanMsg::StorageError(err)).await {
+                    slog::error!(self.logger, "Could not notify control channel of error with APPE: {:?}", err);
+                }
+            }
+        }
+    }
+
     #[tracing_attributes::instrument]
     async fn exec_list_variant(self, path: Option<String>, command: ListCommand) {
         let path = self.resolve_path(path);