From 187c8d6799cf29a58ad9b30dbda180c6cd9b7275 Mon Sep 17 00:00:00 2001 From: Alexander Weiss Date: Wed, 10 Jan 2024 09:18:19 +0100 Subject: [PATCH] add webdav command --- Cargo.toml | 5 + src/commands.rs | 6 +- src/commands/webdav.rs | 84 ++++++++ src/commands/webdav/webdavfs.rs | 327 ++++++++++++++++++++++++++++++++ 4 files changed, 421 insertions(+), 1 deletion(-) create mode 100644 src/commands/webdav.rs create mode 100644 src/commands/webdav/webdavfs.rs diff --git a/Cargo.toml b/Cargo.toml index 3d1c0bb83..aebbbbca0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,6 +89,11 @@ mimalloc = { version = "0.1.39", default_features = false, optional = true } rhai = { workspace = true } simplelog = { workspace = true } runtime-format = "0.1.3" +webdav-handler = {version = "0.2.0", features = ["warp-compat"]} +bytes = "1.5.0" +futures = "0.3.30" +tokio = "1.35.1" +warp = "0.3.6" [dev-dependencies] aho-corasick = { workspace = true } diff --git a/src/commands.rs b/src/commands.rs index 43702c744..ffbaecd3e 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -23,6 +23,7 @@ pub(crate) mod self_update; pub(crate) mod show_config; pub(crate) mod snapshots; pub(crate) mod tag; +pub(crate) mod webdav; use std::fs::File; use std::path::PathBuf; @@ -36,7 +37,7 @@ use crate::{ init::InitCmd, key::KeyCmd, list::ListCmd, ls::LsCmd, merge::MergeCmd, mount::MountCmd, prune::PruneCmd, repair::RepairCmd, repoinfo::RepoInfoCmd, restore::RestoreCmd, self_update::SelfUpdateCmd, show_config::ShowConfigCmd, snapshots::SnapshotCmd, - tag::TagCmd, + tag::TagCmd, webdav::WebDavCmd, }, config::{progress_options::ProgressOptions, RusticConfig}, {Application, RUSTIC_APP}, @@ -129,6 +130,9 @@ enum RusticCmd { /// Change tags of snapshots Tag(TagCmd), + + /// Mount repository + WebDav(WebDavCmd), } /// Entry point for the application. It needs to be a struct to allow using subcommands! diff --git a/src/commands/webdav.rs b/src/commands/webdav.rs new file mode 100644 index 000000000..c8a7d77e2 --- /dev/null +++ b/src/commands/webdav.rs @@ -0,0 +1,84 @@ +//! `mount` subcommand +mod webdavfs; + +use std::net::SocketAddr; + +use crate::{commands::open_repository, status_err, Application, RUSTIC_APP}; +use abscissa_core::{Command, Runnable, Shutdown}; +use anyhow::Result; +use webdav_handler::{warp::dav_handler, DavHandler}; +use webdavfs::RusticWebDavFS; + +#[derive(clap::Parser, Command, Debug)] +pub(crate) struct WebDavCmd { + /// The path template to use for snapshots. {id}, {id_long}, {time}, {username}, {hostname}, {label}, {tags}, {backup_start}, {backup_end} are replaced. [default: "[{hostname}]/[{label}]/{time}"] + #[clap(long)] + path_template: Option, + + /// The time template to use to display times in the path template. See https://docs.rs/chrono/latest/chrono/format/strftime/index.html for format options. [default: "%Y-%m-%d_%H-%M-%S"] + #[clap(long)] + time_template: Option, + + /// Use symlinks. Note this may not be supported by all WebDAV clients + #[clap(long)] + symlinks: bool, + + /// Specify directly which path to mount + #[clap(value_name = "SNAPSHOT[:PATH]")] + snap: Option, +} + +impl Runnable for WebDavCmd { + fn run(&self) { + if let Err(err) = self.inner_run() { + status_err!("{}", err); + RUSTIC_APP.shutdown(Shutdown::Crash); + }; + } +} + +impl WebDavCmd { + fn inner_run(&self) -> Result<()> { + let config = RUSTIC_APP.config(); + + let repo = open_repository(&config)?.to_indexed()?; + + let path_template = self + .path_template + .clone() + .unwrap_or("[{hostname}]/[{label}]/{time}".to_string()); + let time_template = self + .time_template + .clone() + .unwrap_or("%Y-%m-%d_%H-%M-%S".to_string()); + + let sn_filter = |sn: &_| config.snapshot_filter.matches(sn); + let target_fs = if let Some(snap) = &self.snap { + let node = repo.node_from_snapshot_path(snap, sn_filter)?; + RusticWebDavFS::from_node(repo, node) + } else { + let snapshots = repo.get_matching_snapshots(sn_filter)?; + RusticWebDavFS::from_snapshots( + repo, + snapshots, + path_template, + time_template, + self.symlinks, + )? + }; + + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let addr: SocketAddr = ([127, 0, 0, 1], 4918).into(); + + let dav_server = DavHandler::builder().filesystem(target_fs).build_handler(); + + warp::serve(dav_handler(dav_server)).run(addr).await; + }); + + Ok(()) + } +} diff --git a/src/commands/webdav/webdavfs.rs b/src/commands/webdav/webdavfs.rs new file mode 100644 index 000000000..e86f46837 --- /dev/null +++ b/src/commands/webdav/webdavfs.rs @@ -0,0 +1,327 @@ +//! TODO: add blocking() to operation which may block!! + +use std::fmt::{Debug, Formatter}; +use std::io::SeekFrom; +#[cfg(not(windows))] +use std::os::unix::ffi::OsStrExt; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::SystemTime; + +use bytes::{Buf, Bytes}; +use futures::FutureExt; +use rustic_core::repofile::{Node, SnapshotFile}; +use rustic_core::{IndexedFull, OpenFile, Repository}; +use tokio::task; + +use webdav_handler::davpath::DavPath; +use webdav_handler::fs::*; + +use crate::commands::mount::fs::{FsTree, IdenticalSnapshot, Latest}; + +const RUNTIME_TYPE_BASIC: u32 = 1; +const RUNTIME_TYPE_THREADPOOL: u32 = 2; +static RUNTIME_TYPE: AtomicU32 = AtomicU32::new(0); + +fn now() -> SystemTime { + static NOW: OnceLock = OnceLock::new(); + NOW.get_or_init(|| SystemTime::now()).clone() +} + +#[derive(Clone, Copy)] +#[repr(u32)] +enum RuntimeType { + Basic = RUNTIME_TYPE_BASIC, + ThreadPool = RUNTIME_TYPE_THREADPOOL, +} + +impl RuntimeType { + #[inline] + fn get() -> RuntimeType { + match RUNTIME_TYPE.load(Ordering::Relaxed) { + RUNTIME_TYPE_BASIC => RuntimeType::Basic, + RUNTIME_TYPE_THREADPOOL => RuntimeType::ThreadPool, + _ => { + let dbg = format!("{:?}", tokio::runtime::Handle::current()); + let rt = if dbg.contains("ThreadPool") { + RuntimeType::ThreadPool + } else { + RuntimeType::Basic + }; + RUNTIME_TYPE.store(rt as u32, Ordering::SeqCst); + rt + } + } + } +} + +// Run some code via block_in_place() or spawn_blocking(). +#[inline] +async fn blocking(func: F) -> R +where + F: FnOnce() -> R, + F: Send + 'static, + R: Send + 'static, +{ + match RuntimeType::get() { + RuntimeType::Basic => task::spawn_blocking(func).await.unwrap(), + RuntimeType::ThreadPool => task::block_in_place(func), + } +} + +/// DAV Filesystem implementation. +pub struct RusticWebDavFS { + inner: Arc>, +} + +// inner struct. +struct DavFsInner { + repo: Repository, + root: FsTree, +} +impl Debug for DavFsInner { + fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> { + write!(f, "DavFS") + } +} + +struct DavFsFile { + node: Node, + open: OpenFile, + fs: Arc>, + seek: usize, +} +impl Debug for DavFsFile { + fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> { + write!(f, "DavFile") + } +} + +struct DavFsDirEntry(Node); +#[derive(Clone, Debug)] +struct DavFsMetaData(Node); + +impl RusticWebDavFS { + pub(crate) fn from_node(repo: Repository, node: Node) -> Box { + let root = FsTree::RusticTree(node.subtree.unwrap()); + Self::new(repo, root) + } + + pub(crate) fn new(repo: Repository, root: FsTree) -> Box { + let inner = DavFsInner { repo, root }; + Box::new({ + RusticWebDavFS { + inner: Arc::new(inner), + } + }) + } + + pub(crate) fn from_snapshots( + repo: Repository, + snapshots: Vec, + path_template: String, + time_template: String, + symlinks: bool, + ) -> anyhow::Result> { + let (latest, identical) = if symlinks { + (Latest::AsLink, IdenticalSnapshot::AsLink) + } else { + (Latest::AsDir, IdenticalSnapshot::AsDir) + }; + let root = + FsTree::from_snapshots(snapshots, path_template, time_template, latest, identical)?; + Ok(Self::new(repo, root)) + } + + fn node_from_path(&self, path: &DavPath) -> Result { + self.inner + .root + .node_from_path(&self.inner.repo, &path.as_pathbuf()) + .map_err(|_| FsError::GeneralFailure) + } + + fn dir_entries_from_path(&self, path: &DavPath) -> Result, FsError> { + self.inner + .root + .dir_entries_from_path(&self.inner.repo, &path.as_pathbuf()) + .map_err(|_| FsError::GeneralFailure) + } +} + +impl Clone for RusticWebDavFS { + fn clone(&self) -> Self { + RusticWebDavFS { + inner: self.inner.clone(), + } + } +} +// This implementation is basically a bunch of boilerplate to +// wrap the std::fs call in self.blocking() calls. +impl DavFileSystem + for RusticWebDavFS +{ + fn metadata<'a>(&'a self, davpath: &'a DavPath) -> FsFuture<'_, Box> { + self.symlink_metadata(davpath) + } + + fn symlink_metadata<'a>(&'a self, davpath: &'a DavPath) -> FsFuture<'_, Box> { + async move { + let node = self.node_from_path(davpath)?; + Ok(Box::new(DavFsMetaData(node)) as Box) + } + .boxed() + } + + // read_dir is a bit more involved - but not much - than a simple wrapper, + // because it returns a stream. + fn read_dir<'a>( + &'a self, + davpath: &'a DavPath, + _meta: ReadDirMeta, + ) -> FsFuture<'_, FsStream>> { + async move { + let entries = self.dir_entries_from_path(davpath)?; + let v: Vec<_> = entries + .into_iter() + .map(|e| Box::new(DavFsDirEntry(e)) as Box) + .collect(); + let strm = futures::stream::iter(v.into_iter()); + Ok(Box::pin(strm) as FsStream>) + } + .boxed() + } + + fn open<'a>( + &'a self, + path: &'a DavPath, + options: OpenOptions, + ) -> FsFuture<'_, Box> { + async move { + if options.write + || options.append + || options.truncate + || options.create + || options.create_new + { + return Err(FsError::Forbidden); + } + + let node = self.node_from_path(path)?; + let open = self + .inner + .repo + .open_file(&node) + .map_err(|_| FsError::GeneralFailure)?; + Ok(Box::new(DavFsFile { + node, + open, + fs: self.inner.clone(), + seek: 0, + }) as Box) + } + .boxed() + } +} + +impl DavDirEntry for DavFsDirEntry { + fn metadata<'a>(&'a self) -> FsFuture<'_, Box> { + async move { Ok(Box::new(DavFsMetaData(self.0.clone())) as Box) }.boxed() + } + + #[cfg(not(windows))] + fn name(&self) -> Vec { + self.0.name().as_bytes().to_vec() + } + + #[cfg(windows)] + fn name(&self) -> Vec { + self.0 + .name() + .as_os_str() + .to_string_lossy() + .to_string() + .into_bytes() + } +} + +impl DavFile for DavFsFile { + fn metadata<'a>(&'a mut self) -> FsFuture<'_, Box> { + async move { Ok(Box::new(DavFsMetaData(self.node.clone())) as Box) } + .boxed() + } + + fn write_bytes<'a>(&'a mut self, _buf: Bytes) -> FsFuture<'_, ()> { + async move { Err(FsError::Forbidden) }.boxed() + } + + fn write_buf<'a>(&'a mut self, _buf: Box) -> FsFuture<'_, ()> { + async move { Err(FsError::Forbidden) }.boxed() + } + + fn read_bytes<'a>(&'a mut self, count: usize) -> FsFuture<'_, Bytes> { + async move { + let data = self + .fs + .repo + .read_file_at(&self.open, self.seek, count) + .map_err(|_| FsError::GeneralFailure)?; + Ok(data) + } + .boxed() + } + + fn seek<'a>(&'a mut self, pos: SeekFrom) -> FsFuture<'_, u64> { + async move { + match pos { + SeekFrom::Start(start) => self.seek = start as usize, + SeekFrom::Current(delta) => self.seek = (self.seek as i64 + delta) as usize, + SeekFrom::End(end) => self.seek = (self.node.meta.size as i64 + end) as usize, + } + + Ok(self.seek as u64) + } + .boxed() + } + + fn flush<'a>(&'a mut self) -> FsFuture<'_, ()> { + async move { Ok(()) }.boxed() + } +} + +impl DavMetaData for DavFsMetaData { + fn len(&self) -> u64 { + self.0.meta.size + } + fn created(&self) -> FsResult { + Ok(now()) + } + fn modified(&self) -> FsResult { + Ok(self.0.meta.mtime.map(SystemTime::from).unwrap_or(now())) + } + fn accessed(&self) -> FsResult { + Ok(self.0.meta.atime.map(SystemTime::from).unwrap_or(now())) + } + + fn status_changed(&self) -> FsResult { + Ok(self.0.meta.ctime.map(SystemTime::from).unwrap_or(now())) + } + + fn is_dir(&self) -> bool { + self.0.is_dir() + } + fn is_file(&self) -> bool { + self.0.is_file() + } + fn is_symlink(&self) -> bool { + self.0.is_symlink() + } + fn executable(&self) -> FsResult { + if self.0.is_file() { + let Some(mode) = self.0.meta.mode else { + return Ok(false); + }; + return Ok((mode & 0o100) > 0); + } + Err(FsError::NotImplemented) + } +}