ftp.rs
raw
//! FTP backend.
//!
//! Repository objects are already sealed client-side, so plain FTP only
//! exposes what any storage host sees anyway; the notable residual risk
//! is that classic FTP sends the *credentials* in the clear, so anyone
//! on the wire can gain the access needed to delete backups.
//!
//! URL form: `ftp://user:password@host:port/path`. Credentials are
//! resolved most-specific first: the URL itself, `$BEEPING_FTP_PASSWORD`,
//! then a matching lftp bookmark (see [`crate::lftp`]); with nothing
//! found, the login is `anonymous` with an empty password — plenty of
//! NAS boxes serve exactly that. Writing `ftp://anonymous@host/...`
//! forces anonymous login even when a bookmark holds named credentials
//! for the host.
//!
//! At most `?connections=N` (default 4) connections are held at once,
//! and a server that refuses further logins (421) is waited out rather
//! than treated as an error — see [`crate::pool`].
use std::sync::{
Mutex,
atomic::{AtomicBool, Ordering},
};
use repository::{Backend, ChunkId, Error, ObjectKey, ObjectKind};
use suppaftp::{FtpError, FtpStream, Status, types::FileType};
use url::Url;
use crate::{
lftp, percent_decode,
pool::{Pool, temp_suffix},
};
const DEFAULT_CONNECTIONS: usize = 4;
pub struct FtpBackend {
host: String,
port: u16,
user: String,
password: String,
root: String,
pool: Pool<FtpStream>,
/// Directories already created this session, to avoid a MKD round
/// trip per stored object.
known_dirs: Mutex<std::collections::HashSet<String>>,
/// Set once the server has declined to list machine-readably, which
/// is the only way to learn what the repository occupies. Asking
/// again would cost a retry and a fresh dial for the same refusal.
size_unknown: AtomicBool,
}
impl FtpBackend {
pub fn from_url(url: &Url) -> Result<FtpBackend, Error> {
let host = url
.host_str()
.ok_or_else(|| Error::Backend(format!("{url} has no host")))?
.to_string();
let port = url.port().unwrap_or(21);
let mut user = match url.username() {
"" => None,
user => Some(percent_decode(user)),
};
let mut password = url.password().map(percent_decode);
if password.is_none()
&& let Ok(env) = std::env::var("BEEPING_FTP_PASSWORD")
{
password = Some(env);
}
if (user.is_none() || password.is_none())
&& let Some(found) = lftp::find_credentials("ftp", &host, port, user.as_deref())
{
user.get_or_insert(found.user);
if password.is_none() {
password = found.password;
}
}
let user = user.unwrap_or_else(|| "anonymous".to_string());
let password = password.unwrap_or_default();
Ok(FtpBackend {
host,
port,
user,
password,
root: url.path().trim_end_matches('/').to_string(),
pool: Pool::new(crate::connection_limit(url, DEFAULT_CONNECTIONS)?),
size_unknown: AtomicBool::new(false),
known_dirs: Mutex::new(std::collections::HashSet::new()),
})
}
fn with_connection<T>(
&self,
op: impl FnMut(&mut FtpStream) -> Result<T, Error>,
) -> Result<T, Error> {
self.pool.with(|| self.connect(), is_server_full, op)
}
fn connect(&self) -> Result<FtpStream, Error> {
let mut ftp = FtpStream::connect((self.host.as_str(), self.port)).map_err(ftp_error)?;
ftp.login(&self.user, &self.password).map_err(ftp_error)?;
ftp.transfer_type(FileType::Binary).map_err(ftp_error)?;
Ok(ftp)
}
fn object_path(&self, key: &ObjectKey) -> String {
let mut path = self.root.clone();
for segment in key.segments() {
path.push('/');
path.push_str(&segment);
}
path
}
fn dir_path(&self, segments: &[&str]) -> String {
let mut path = self.root.clone();
for segment in segments {
path.push('/');
path.push_str(segment);
}
path
}
/// Creates the directories leading to `key` — the repository root
/// and every intermediate — remembering successes. MKD failures are
/// ignored here: "already exists" is not reliably distinguishable
/// in FTP, and a real problem surfaces as a failed STOR immediately
/// after.
fn ensure_parents(&self, ftp: &mut FtpStream, key: &ObjectKey) {
let segments = key.segments();
let mut directories = Vec::new();
if !self.root.is_empty() {
directories.push(self.root.clone());
}
let mut path = self.root.clone();
// All but the last segment are directories
for segment in &segments[..segments.len() - 1] {
path.push('/');
path.push_str(segment);
directories.push(path.clone());
}
for directory in directories {
let mut known = self.known_dirs.lock().unwrap();
if known.contains(&directory) {
continue;
}
let _ = ftp.mkdir(&directory);
known.insert(directory);
}
}
}
impl Backend for FtpBackend {
fn put(&self, key: &ObjectKey, data: &[u8]) -> Result<(), Error> {
let path = self.object_path(key);
self.with_connection(|ftp| {
if ftp.size(&path).is_ok() {
// Objects are write-once; it is already this one
return Ok(());
}
self.ensure_parents(ftp, key);
// Upload beside the final name, then rename into place
// so no reader (or crash) ever sees a partial object
let temp = format!("{path}{}", temp_suffix());
ftp.put_file(&temp, &mut std::io::Cursor::new(data))
.map_err(ftp_error)?;
match ftp.rename(&temp, &path) {
Ok(()) => Ok(()),
Err(_) if ftp.size(&path).is_ok() => {
// A concurrent writer beat us to it; ours is
// surplus
let _ = ftp.rm(&temp);
Ok(())
}
Err(err) => Err(ftp_error(err)),
}
})
}
fn get(&self, key: &ObjectKey) -> Result<Option<Vec<u8>>, Error> {
let path = self.object_path(key);
self.with_connection(|ftp| match ftp.retr_as_buffer(&path) {
Ok(buffer) => Ok(Some(buffer.into_inner())),
Err(err) if is_not_found(&err) => Ok(None),
Err(err) => Err(ftp_error(err)),
})
}
fn contains(&self, key: &ObjectKey) -> Result<bool, Error> {
let path = self.object_path(key);
self.with_connection(|ftp| match ftp.size(&path) {
Ok(_) => Ok(true),
Err(err) if is_not_found(&err) => Ok(false),
Err(err) => Err(ftp_error(err)),
})
}
fn list(
&self,
kind: ObjectKind,
visit: &mut dyn FnMut(ObjectKey) -> Result<(), Error>,
) -> Result<(), Error> {
match kind {
ObjectKind::Header => {
if self.contains(&ObjectKey::Header)? {
visit(ObjectKey::Header)?;
}
}
// The flat kinds: one directory of hex-named objects each
ObjectKind::Snapshot | ObjectKind::Prune | ObjectKind::Lock => {
for name in self.dir_names(&self.dir_path(&[kind.directory()]))? {
if let Some(key) = kind.key_for(&name) {
visit(key)?;
}
}
}
ObjectKind::Chunk => {
for prefix in self.dir_names(&self.dir_path(&["chunks"]))? {
if prefix.len() != 2 || !prefix.bytes().all(|b| b.is_ascii_hexdigit()) {
continue;
}
for name in self.dir_names(&self.dir_path(&["chunks", &prefix]))? {
if let Ok(id) = ChunkId::from_hex(name) {
visit(ObjectKey::Chunk(id))?;
}
}
}
}
}
Ok(())
}
/// Adds up everything under the repository directory, from MLSD
/// listings — one per directory, so a few hundred round trips for
/// the fanned-out chunk store. Asked once a run, never on a hot
/// path.
///
/// MLSD is the only listing FTP has whose sizes can be read without
/// guessing at a server's chosen `ls` format. A server without it
/// cannot say what it holds, and says so once rather than being
/// asked again.
fn used_space(&self) -> Result<Option<u64>, Error> {
if self.size_unknown.load(Ordering::Acquire) {
return Ok(None);
}
let root = self.root.clone();
match self.with_connection(|ftp| tree_bytes(ftp, &root)) {
Ok(total) => Ok(Some(total)),
Err(_) => {
self.size_unknown.store(true, Ordering::Release);
Ok(None)
}
}
}
fn delete(&self, key: &ObjectKey) -> Result<(), Error> {
let path = self.object_path(key);
self.with_connection(|ftp| match ftp.rm(&path) {
Ok(()) => Ok(()),
Err(err) if is_not_found(&err) => Ok(()),
Err(err) => Err(ftp_error(err)),
})
}
}
impl FtpBackend {
/// The entry names within a directory, treating a missing directory
/// as empty. Servers differ on whether NLST returns bare names or
/// full paths, so paths are reduced to their final component.
fn dir_names(&self, path: &str) -> Result<Vec<String>, Error> {
self.with_connection(|ftp| match ftp.nlst(Some(path)) {
Ok(names) => Ok(names
.into_iter()
.map(|name| {
name.rsplit('/')
.next()
.expect("rsplit always yields at least one part")
.to_string()
})
.collect()),
Err(err) if is_not_found(&err) => Ok(Vec::new()),
Err(err) => Err(ftp_error(err)),
})
}
}
/// How many bytes the tree under `path` holds, by the sizes MLSD
/// reports. A directory that is not there holds nothing — the
/// repository's own subdirectories appear only once something has been
/// put in them.
fn tree_bytes(ftp: &mut FtpStream, path: &str) -> Result<u64, Error> {
let lines = match ftp.mlsd(Some(path)) {
Ok(lines) => lines,
Err(err) if is_not_found(&err) => return Ok(0),
Err(err) => return Err(ftp_error(err)),
};
let mut total = 0;
for line in lines {
let Some(listed) = Listed::parse(&line) else {
continue;
};
total += match listed.kind {
// Servers differ on whether the name is bare or a full
// path, exactly as they do for NLST
ListedKind::Directory => tree_bytes(ftp, &format!("{path}/{}", listed.name))?,
ListedKind::File => listed.size,
};
}
Ok(total)
}
/// One entry of an MLSD listing: `fact=value;...; name`, per RFC 3659.
struct Listed<'a> {
kind: ListedKind,
size: u64,
name: &'a str,
}
enum ListedKind {
File,
Directory,
}
impl<'a> Listed<'a> {
/// Reads one listing line, or `None` for the entries that are not
/// objects of their own: the directory itself, its parent, and
/// anything whose type the server did not give.
fn parse(line: &'a str) -> Option<Listed<'a>> {
let (facts, name) = line.trim_end().split_once(' ')?;
let mut kind = None;
let mut size = 0;
for fact in facts.split(';').filter(|fact| !fact.is_empty()) {
let Some((key, value)) = fact.split_once('=') else {
continue;
};
match key.to_ascii_lowercase().as_str() {
"type" => {
kind = match value.to_ascii_lowercase().as_str() {
"file" => Some(ListedKind::File),
"dir" => Some(ListedKind::Directory),
// cdir and pdir are this directory and its
// parent, listed for navigation
_ => return None,
};
}
"size" => size = value.parse().unwrap_or(0),
_ => {}
}
}
Some(Listed {
kind: kind?,
size,
name: name.rsplit('/').next().expect("rsplit yields a part"),
})
}
}
/// FTP reports a missing file as 550; the code is also used for
/// permission problems, but that ambiguity is inherent to the protocol.
fn is_not_found(err: &FtpError) -> bool {
matches!(
err,
FtpError::UnexpectedResponse(response) if response.status == Status::FileUnavailable
)
}
/// Maps FTP errors into backend errors, giving 421 — the server
/// declining service, most commonly "too many users already logged
/// in" — its own variant so the connection pool can wait it out rather
/// than fail.
fn ftp_error(err: FtpError) -> Error {
if matches!(
&err,
FtpError::UnexpectedResponse(response) if response.status == Status::NotAvailable
) {
return Error::BackendBusy(format!("ftp: {err}"));
}
Error::Backend(format!("ftp: {err}"))
}
fn is_server_full(err: &Error) -> bool {
matches!(err, Error::BackendBusy(_))
}
#[cfg(test)]
mod tests {
use super::*;
// These use a fictional host so no real lftp bookmark can match,
// and they tolerate $BEEPING_FTP_PASSWORD being set in the
// developer's environment rather than mutating process env.
#[test]
fn credentialless_urls_default_to_anonymous() {
let url = Url::parse("ftp://nas.invalid/backups/repo").unwrap();
let backend = FtpBackend::from_url(&url).unwrap();
assert_eq!(backend.user, "anonymous");
if std::env::var("BEEPING_FTP_PASSWORD").is_err() {
assert_eq!(backend.password, "");
}
}
#[test]
fn an_explicit_anonymous_user_gets_an_empty_password() {
let url = Url::parse("ftp://anonymous@nas.invalid/backups/repo").unwrap();
let backend = FtpBackend::from_url(&url).unwrap();
assert_eq!(backend.user, "anonymous");
if std::env::var("BEEPING_FTP_PASSWORD").is_err() {
assert_eq!(backend.password, "");
}
}
#[test]
fn url_credentials_pass_through_decoded() {
let url = Url::parse("ftp://alice:p%40ss@nas.invalid:2121/repo").unwrap();
let backend = FtpBackend::from_url(&url).unwrap();
assert_eq!(backend.user, "alice");
assert_eq!(backend.password, "p@ss");
assert_eq!(backend.port, 2121);
assert_eq!(backend.root, "/repo");
}
/// MLSD is the one FTP listing whose sizes can be read without
/// guessing at a server's `ls` format, and the facts it carries are
/// unordered, case-insensitive, and open-ended.
#[test]
fn mlsd_lines_give_up_their_sizes() {
let file = Listed::parse("type=file;size=1048576;modify=20260823090000; ab12cd")
.expect("a file entry");
assert!(matches!(file.kind, ListedKind::File));
assert_eq!(file.size, 1048576);
assert_eq!(file.name, "ab12cd");
// Order and case are the server's business, and unknown facts
// are to be ignored rather than tripped over
let directory =
Listed::parse("perm=fle;Type=dir;unique=12U1; chunks").expect("a directory");
assert!(matches!(directory.kind, ListedKind::Directory));
assert_eq!(directory.name, "chunks");
// Some servers answer with full paths, as they do for NLST
assert_eq!(
Listed::parse("type=dir; /backups/repo/chunks/00")
.expect("a directory")
.name,
"00"
);
// A name may hold spaces; only the first one ends the facts
assert_eq!(
Listed::parse("type=file;size=3; two words")
.expect("a file")
.name,
"two words"
);
// This directory and its parent are listed for navigation, and
// adding them up would count the tree twice over
assert!(Listed::parse("type=cdir;modify=20260823090000; .").is_none());
assert!(Listed::parse("type=pdir; ..").is_none());
// Nothing to make of these
assert!(Listed::parse("size=12; no-type").is_none());
assert!(Listed::parse("").is_none());
assert!(Listed::parse("type=file;size=7;").is_none());
}
}