memona-plugin-ftp/src/control.rs

//! The FTP control channel: one socket, line replies, and passive transfers.

use memona_plugin_sdk::ErrorCode;
use memona_plugin_sdk::memona::filesystem::host;

const MAX_LINE: usize = 8192;
const MAX_REPLY_LINES: usize = 256;

pub fn transport(error: host::HostError) -> ErrorCode {
  match error {
    host::HostError::Denied => ErrorCode::Denied,
    host::HostError::Cancelled => ErrorCode::Cancelled,
    host::HostError::InvalidInput => ErrorCode::InvalidInput,
    host::HostError::Internal => ErrorCode::Internal,
    host::HostError::Unavailable => ErrorCode::Unavailable,
  }
}

/// What an FTP reply code means to a filesystem caller.
pub fn outcome(code: u16) -> Result<(), ErrorCode> {
  match code {
    200..=399 => Ok(()),
    // 530 is a refused login and 550 covers both a missing file and a denied
    // one; the server does not distinguish them, so neither does this.
    530 | 532 => Err(ErrorCode::Denied),
    550 => Err(ErrorCode::NotFound),
    450 | 452 | 552 => Err(ErrorCode::Conflict),
    502 | 504 => Err(ErrorCode::Unsupported),
    _ => Err(ErrorCode::Unavailable),
  }
}

pub struct Control {
  socket: host::Tcp,
  host: String,
  buffer: Vec<u8>,
}

impl Control {
  pub fn open(
    host_name: String,
    port: u16,
    tls: bool,
  ) -> Result<Self, ErrorCode> {
    let socket = host::connect(&host_name, port, tls).map_err(transport)?;
    let mut control = Self {
      socket,
      host: host_name,
      buffer: Vec::new(),
    };
    // The greeting is a reply like any other, and a 4xx or 5xx one means the
    // server is refusing the session before a command was even sent.
    let (code, _) = control.reply()?;
    outcome(code)?;
    Ok(control)
  }

  fn line(&mut self) -> Result<String, ErrorCode> {
    loop {
      if let Some(index) = self.buffer.iter().position(|byte| *byte == b'\n') {
        let line: Vec<u8> = self.buffer.drain(..=index).collect();
        return Ok(
          String::from_utf8_lossy(&line)
            .trim_end_matches(['\r', '\n'])
            .to_owned(),
        );
      }
      if self.buffer.len() > MAX_LINE {
        return Err(ErrorCode::Unavailable);
      }
      let chunk = self.socket.read(4096).map_err(transport)?;
      if chunk.is_empty() {
        return Err(ErrorCode::Unavailable);
      }
      self.buffer.extend_from_slice(&chunk);
    }
  }

  /// Reads one complete reply, following the `NNN-` continuation form.
  pub fn reply(&mut self) -> Result<(u16, String), ErrorCode> {
    let first = self.line()?;
    let code: u16 = first
      .get(..3)
      .and_then(|text| text.parse().ok())
      .ok_or(ErrorCode::Unavailable)?;
    if first.as_bytes().get(3) != Some(&b'-') {
      return Ok((code, first));
    }
    let terminator = format!("{code} ");
    for _ in 0..MAX_REPLY_LINES {
      let line = self.line()?;
      if line.starts_with(&terminator) {
        return Ok((code, line));
      }
    }
    Err(ErrorCode::Unavailable)
  }

  /// Sends one command. A newline in an argument would forge a second command,
  /// so it is refused rather than escaped.
  pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
    if command.contains(['\r', '\n']) || command.len() > MAX_LINE {
      return Err(ErrorCode::InvalidInput);
    }
    let line = format!("{command}\r\n");
    let mut sent = 0;
    while sent < line.len() {
      let written = self
        .socket
        .write(&line.as_bytes()[sent..])
        .map_err(transport)? as usize;
      if written == 0 {
        return Err(ErrorCode::Unavailable);
      }
      sent += written;
    }
    self.reply()
  }

  pub fn expect(&mut self, command: &str) -> Result<String, ErrorCode> {
    let (code, text) = self.send(command)?;
    outcome(code)?;
    Ok(text)
  }

  /// Opens a passive data connection, preferring EPSV so the address the
  /// server names is never one this plugin could be redirected to.
  pub fn passive(&mut self) -> Result<host::Tcp, ErrorCode> {
    let text = self.expect("EPSV")?;
    let port = extended_port(&text).ok_or(ErrorCode::Unavailable)?;
    host::connect(&self.host, port, false).map_err(transport)
  }
}

/// Reads the port out of `229 Entering Extended Passive Mode (|||1234|)`.
fn extended_port(text: &str) -> Option<u16> {
  let start = text.find('(')?;
  let end = text[start..].find(')')? + start;
  let inner = text.get(start + 1..end)?;
  let separator = inner.chars().next()?;
  inner
    .rsplit(separator)
    .find(|part| !part.is_empty())?
    .parse()
    .ok()
}

#[cfg(test)]
mod tests {
  use super::{extended_port, outcome};
  use memona_plugin_sdk::ErrorCode;

  #[test]
  fn reads_the_port_from_an_extended_passive_reply() {
    assert_eq!(
      extended_port("229 Entering Extended Passive Mode (|||49152|)"),
      Some(49152)
    );
    // The separator is whatever the server chose, not always a pipe.
    assert_eq!(extended_port("229 ok (!!!1024!)"), Some(1024));
    assert_eq!(extended_port("229 ok (|||0|)"), Some(0));
    assert_eq!(extended_port("229 ok"), None);
    assert_eq!(extended_port("229 ok (|||not-a-port|)"), None);
  }

  #[test]
  fn reply_codes_keep_refused_and_missing_apart() {
    assert!(outcome(226).is_ok());
    assert!(matches!(outcome(530), Err(ErrorCode::Denied)));
    assert!(matches!(outcome(550), Err(ErrorCode::NotFound)));
    assert!(matches!(outcome(452), Err(ErrorCode::Conflict)));
    assert!(matches!(outcome(502), Err(ErrorCode::Unsupported)));
    assert!(matches!(outcome(421), Err(ErrorCode::Unavailable)));
  }
}