src/tools/fuse_loop.rs

   1 //! Map a raw data reader as a loop device via FUSE
   2
   3 use anyhow::{Error, format_err, bail};
   4 use std::ffi::OsStr;
   5 use std::path::{Path, PathBuf};
   6 use std::fs::{File, remove_file, read_to_string, OpenOptions};
   7 use std::io::SeekFrom;
   8 use std::io::prelude::*;
   9 use std::collections::HashMap;
  10
  11 use nix::unistd::Pid;
  12 use nix::sys::signal::{self, Signal};
  13
  14 use tokio::io::{AsyncRead, AsyncSeek, AsyncReadExt, AsyncSeekExt};
  15 use futures::stream::{StreamExt, TryStreamExt};
  16 use futures::channel::mpsc::{Sender, Receiver};
  17
  18 use proxmox::const_regex;
  19 use proxmox::tools::time;
  20 use proxmox_fuse::{*, requests::FuseRequest};
  21 use super::loopdev;
  22
  23 const RUN_DIR: &str = "/run/pbs-loopdev";
  24
  25 const_regex! {
  26     pub LOOPDEV_REGEX = r"^loop\d+$";
  27 }
  28
  29 /// Represents an ongoing FUSE-session that has been mapped onto a loop device.
  30 /// Create with map_loop, then call 'main' and poll until startup_chan reports
  31 /// success. Then, daemonize or otherwise finish setup, and continue polling
  32 /// main's future until completion.
  33 pub struct FuseLoopSession<R: AsyncRead + AsyncSeek + Unpin> {
  34     session: Option<Fuse>,
  35     stat: libc::stat,
  36     reader: R,
  37     fuse_path: String,
  38     pid_path: String,
  39     pub loopdev_path: String,
  40 }
  41
  42 impl<R: AsyncRead + AsyncSeek + Unpin> FuseLoopSession<R> {
  43
  44     /// Prepare for mapping the given reader as a block device node at
  45     /// /dev/loopN. Creates a temporary file for FUSE and a PID file for unmap.
  46     pub async fn map_loop<P: AsRef<str>>(size: u64, mut reader: R, name: P, options: &OsStr)
  47         -> Result<Self, Error>
  48     {
  49         // attempt a single read to check if the reader is configured correctly
  50         let _ = reader.read_u8().await?;
  51
  52         std::fs::create_dir_all(RUN_DIR)?;
  53         let mut path = PathBuf::from(RUN_DIR);
  54         path.push(name.as_ref());
  55         let mut pid_path = path.clone();
  56         pid_path.set_extension("pid");
  57
  58         // cleanup previous instance with same name
  59         // if loopdev is actually still mapped, this will do nothing and the
  60         // create_new below will fail as intended
  61         cleanup_unused_run_files(Some(name.as_ref().to_owned()));
  62
  63         match OpenOptions::new().write(true).create_new(true).open(&path) {
  64             Ok(_) => { /* file created, continue on */ },
  65             Err(e) => {
  66                 if e.kind() == std::io::ErrorKind::AlreadyExists {
  67                     bail!("the given archive is already mapped, cannot map twice");
  68                 } else {
  69                     bail!("error while creating backing file ({:?}) - {}", &path, e);
  70                 }
  71             },
  72         }
  73
  74         let session = Fuse::builder("pbs-block-dev")?
  75             .options_os(options)?
  76             .enable_read()
  77             .build()?
  78             .mount(&path)?;
  79
  80         let loopdev_path = loopdev::get_or_create_free_dev().map_err(|err| {
  81             format_err!("loop-control GET_FREE failed - {}", err)
  82         })?;
  83
  84         // write pidfile so unmap can later send us a signal to exit
  85         Self::write_pidfile(&pid_path)?;
  86
  87         Ok(Self {
  88             session: Some(session),
  89             reader,
  90             stat: minimal_stat(size as i64),
  91             fuse_path: path.to_string_lossy().into_owned(),
  92             pid_path: pid_path.to_string_lossy().into_owned(),
  93             loopdev_path,
  94         })
  95     }
  96
  97     fn write_pidfile(path: &Path) -> Result<(), Error> {
  98         let pid = unsafe { libc::getpid() };
  99         let mut file = File::create(path)?;
 100         write!(file, "{}", pid)?;
 101         Ok(())
 102     }
 103
 104     /// Runs the FUSE request loop and assigns the loop device. Will send a
 105     /// message on startup_chan once the loop device is assigned (or assignment
 106     /// fails). Send a message on abort_chan to trigger cleanup and exit FUSE.
 107     /// An error on loopdev assignment does *not* automatically close the FUSE
 108     /// handle or do cleanup, trigger abort_chan manually in case startup fails.
 109     pub async fn main(
 110         &mut self,
 111         mut startup_chan: Sender<Result<(), Error>>,
 112         abort_chan: Receiver<()>,
 113     ) -> Result<(), Error> {
 114
 115         if self.session.is_none() {
 116             panic!("internal error: fuse_loop::main called before ::map_loop");
 117         }
 118         let mut session = self.session.take().unwrap().fuse();
 119         let mut abort_chan = abort_chan.fuse();
 120
 121         let (loopdev_path, fuse_path) = (self.loopdev_path.clone(), self.fuse_path.clone());
 122         tokio::task::spawn_blocking(move || {
 123             if let Err(err) = loopdev::assign(loopdev_path, fuse_path) {
 124                 let _ = startup_chan.try_send(Err(format_err!("error while assigning loop device - {}", err)));
 125             } else {
 126                 // device is assigned successfully, which means not only is the
 127                 // loopdev ready, but FUSE is also okay, since the assignment
 128                 // would have failed otherwise
 129                 let _ = startup_chan.try_send(Ok(()));
 130             }
 131         });
 132
 133         let (loopdev_path, fuse_path, pid_path) =
 134             (self.loopdev_path.clone(), self.fuse_path.clone(), self.pid_path.clone());
 135         let cleanup = |session: futures::stream::Fuse<Fuse>| {
 136             // only warn for errors on cleanup, if these fail nothing is lost
 137             if let Err(err) = loopdev::unassign(&loopdev_path) {
 138                 eprintln!(
 139                     "cleanup: warning: could not unassign file {} from loop device {} - {}",
 140                     &fuse_path,
 141                     &loopdev_path,
 142                     err,
 143                 );
 144             }
 145
 146             // force close FUSE handle before attempting to remove backing file
 147             std::mem::drop(session);
 148
 149             if let Err(err) = remove_file(&fuse_path) {
 150                 eprintln!(
 151                     "cleanup: warning: could not remove temporary file {} - {}",
 152                     &fuse_path,
 153                     err,
 154                 );
 155             }
 156             if let Err(err) = remove_file(&pid_path) {
 157                 eprintln!(
 158                     "cleanup: warning: could not remove PID file {} - {}",
 159                     &pid_path,
 160                     err,
 161                 );
 162             }
 163         };
 164
 165         loop {
 166             tokio::select!{
 167                 _ = abort_chan.next() => {
 168                     // aborted, do cleanup and exit
 169                     break;
 170                 },
 171                 req = session.try_next() => {
 172                     let res = match req? {
 173                         Some(Request::Lookup(req)) => {
 174                             let stat = self.stat;
 175                             let entry = EntryParam::simple(stat.st_ino, stat);
 176                             req.reply(&entry)
 177                         },
 178                         Some(Request::Getattr(req)) => {
 179                             req.reply(&self.stat, std::f64::MAX)
 180                         },
 181                         Some(Request::Read(req)) => {
 182                             match self.reader.seek(SeekFrom::Start(req.offset)).await {
 183                                 Ok(_) => {
 184                                     let mut buf = vec![0u8; req.size];
 185                                     match self.reader.read_exact(&mut buf).await {
 186                                         Ok(_) => {
 187                                             req.reply(&buf)
 188                                         },
 189                                         Err(e) => {
 190                                             req.io_fail(e)
 191                                         }
 192                                     }
 193                                 },
 194                                 Err(e) => {
 195                                     req.io_fail(e)
 196                                 }
 197                             }
 198                         },
 199                         Some(_) => {
 200                             // only FUSE requests necessary for loop-mapping are implemented
 201                             eprintln!("Unimplemented FUSE request type encountered");
 202                             Ok(())
 203                         },
 204                         None => {
 205                             // FUSE connection closed
 206                             break;
 207                         }
 208                     };
 209                     if let Err(err) = res {
 210                         // error during FUSE reply, cleanup and exit
 211                         cleanup(session);
 212                         bail!(err);
 213                     }
 214                 }
 215             }
 216         }
 217
 218         // non-error FUSE exit
 219         cleanup(session);
 220         Ok(())
 221     }
 222 }
 223
 224 /// Clean up leftover files as well as FUSE instances without a loop device
 225 /// connected. Best effort, never returns an error.
 226 /// If filter_name is Some("..."), only this name will be cleaned up.
 227 pub fn cleanup_unused_run_files(filter_name: Option<String>) {
 228     if let Ok(maps) = find_all_mappings() {
 229         for (name, loopdev) in maps {
 230             if loopdev.is_none() &&
 231                 (filter_name.is_none() || &name == filter_name.as_ref().unwrap())
 232             {
 233                 let mut path = PathBuf::from(RUN_DIR);
 234                 path.push(&name);
 235
 236                 // clean leftover FUSE instances (e.g. user called 'losetup -d' or similar)
 237                 // does nothing if files are already stagnant (e.g. instance crashed etc...)
 238                 if unmap_from_backing(&path, None).is_ok() {
 239                     // we have reaped some leftover instance, tell the user
 240                     eprintln!(
 241                         "Cleaned up dangling mapping '{}': no loop device assigned",
 242                         &name
 243                     );
 244                 }
 245
 246                 // remove remnant files
 247                 // these we're not doing anything, so no need to inform the user
 248                 let _ = remove_file(&path);
 249                 path.set_extension("pid");
 250                 let _ = remove_file(&path);
 251             }
 252         }
 253     }
 254 }
 255
 256 fn get_backing_file(loopdev: &str) -> Result<String, Error> {
 257     let num = loopdev.split_at(9).1.parse::<u8>().map_err(|err|
 258         format_err!("malformed loopdev path, does not end with valid number - {}", err))?;
 259
 260     let block_path = PathBuf::from(format!("/sys/devices/virtual/block/loop{}/loop/backing_file", num));
 261     let backing_file = read_to_string(block_path).map_err(|err| {
 262         if err.kind() == std::io::ErrorKind::NotFound {
 263             format_err!("nothing mapped to {}", loopdev)
 264         } else {
 265             format_err!("error reading backing file - {}", err)
 266         }
 267     })?;
 268
 269     let backing_file = backing_file.trim();
 270
 271     if !backing_file.starts_with(RUN_DIR) {
 272         bail!(
 273             "loopdev {} is in use, but not by proxmox-backup-client (mapped to '{}')",
 274             loopdev,
 275             backing_file,
 276         );
 277     }
 278
 279     Ok(backing_file.to_owned())
 280 }
 281
 282 // call in broken state: we found the mapping, but the client is already dead,
 283 // only thing to do is clean up what we can
 284 fn emerg_cleanup (loopdev: Option<&str>, mut backing_file: PathBuf) {
 285     eprintln!(
 286         "warning: found mapping with dead process ({:?}), attempting cleanup",
 287         &backing_file
 288     );
 289
 290     if let Some(loopdev) = loopdev {
 291         let _ = loopdev::unassign(loopdev);
 292     }
 293
 294     // killing the backing process does not cancel the FUSE mount automatically
 295     let mut command = std::process::Command::new("fusermount");
 296     command.arg("-u");
 297     command.arg(&backing_file);
 298     let _ = crate::tools::run_command(command, None);
 299
 300     let _ = remove_file(&backing_file);
 301     backing_file.set_extension("pid");
 302     let _ = remove_file(&backing_file);
 303 }
 304
 305 fn unmap_from_backing(backing_file: &Path, loopdev: Option<&str>) -> Result<(), Error> {
 306     let mut pid_path = PathBuf::from(backing_file);
 307     pid_path.set_extension("pid");
 308
 309     let pid_str = read_to_string(&pid_path).map_err(|err| {
 310         if err.kind() == std::io::ErrorKind::NotFound {
 311             emerg_cleanup(loopdev, backing_file.to_owned());
 312         }
 313         format_err!("error reading pidfile {:?}: {}", &pid_path, err)
 314     })?;
 315     let pid = pid_str.parse::<i32>().map_err(|err|
 316         format_err!("malformed PID ({}) in pidfile - {}", pid_str, err))?;
 317
 318     let pid = Pid::from_raw(pid);
 319
 320     // send SIGINT to trigger cleanup and exit in target process
 321     match signal::kill(pid, Signal::SIGINT) {
 322         Ok(()) => {},
 323         Err(nix::Error::Sys(nix::errno::Errno::ESRCH)) => {
 324             emerg_cleanup(loopdev, backing_file.to_owned());
 325             return Ok(());
 326         },
 327         Err(e) => return Err(e.into()),
 328     }
 329
 330     // block until unmap is complete or timeout
 331     let start = time::epoch_i64();
 332     loop {
 333         match signal::kill(pid, None) {
 334             Ok(_) => {
 335                 // 10 second timeout, then assume failure
 336                 if (time::epoch_i64() - start) > 10 {
 337                     return Err(format_err!("timed out waiting for PID '{}' to exit", &pid));
 338                 }
 339                 std::thread::sleep(std::time::Duration::from_millis(100));
 340             },
 341             Err(nix::Error::Sys(nix::errno::Errno::ESRCH)) => {
 342                 break;
 343             },
 344             Err(e) => return Err(e.into()),
 345         }
 346     }
 347
 348     Ok(())
 349 }
 350
 351 /// Returns an Iterator over a set of currently active mappings, i.e.
 352 /// FuseLoopSession instances. Returns ("backing-file-name", Some("/dev/loopX"))
 353 /// where .1 is None when a user has manually called 'losetup -d' or similar but
 354 /// the FUSE instance is still running.
 355 pub fn find_all_mappings() -> Result<impl Iterator<Item = (String, Option<String>)>, Error> {
 356     // get map of all /dev/loop mappings belonging to us
 357     let mut loopmap = HashMap::new();
 358     for ent in pbs_tools::fs::scan_subdir(libc::AT_FDCWD, Path::new("/dev/"), &LOOPDEV_REGEX)? {
 359         if let Ok(ent) = ent {
 360             let loopdev = format!("/dev/{}", ent.file_name().to_string_lossy());
 361             if let Ok(file) = get_backing_file(&loopdev) {
 362                 // insert filename only, strip RUN_DIR/
 363                 loopmap.insert(file[RUN_DIR.len()+1..].to_owned(), loopdev);
 364             }
 365         }
 366     }
 367
 368     Ok(pbs_tools::fs::read_subdir(libc::AT_FDCWD, Path::new(RUN_DIR))?
 369         .filter_map(move |ent| {
 370             match ent {
 371                 Ok(ent) => {
 372                     let file = ent.file_name().to_string_lossy();
 373                     if file == "." || file == ".." || file.ends_with(".pid") {
 374                         None
 375                     } else {
 376                         let loopdev = loopmap.get(file.as_ref()).map(String::to_owned);
 377                         Some((file.into_owned(), loopdev))
 378                     }
 379                 },
 380                 Err(_) => None,
 381             }
 382         }))
 383 }
 384
 385 /// Try and unmap a running proxmox-backup-client instance from the given
 386 /// /dev/loopN device
 387 pub fn unmap_loopdev<S: AsRef<str>>(loopdev: S) -> Result<(), Error> {
 388     let loopdev = loopdev.as_ref();
 389     if loopdev.len() < 10 || !loopdev.starts_with("/dev/loop") {
 390         bail!("malformed loopdev path, must be in format '/dev/loopX'");
 391     }
 392
 393     let backing_file = get_backing_file(loopdev)?;
 394     unmap_from_backing(Path::new(&backing_file), Some(loopdev))
 395 }
 396
 397 /// Try and unmap a running proxmox-backup-client instance from the given name
 398 pub fn unmap_name<S: AsRef<str>>(name: S) -> Result<(), Error> {
 399     for (mapping, loopdev) in find_all_mappings()? {
 400         if mapping.ends_with(name.as_ref()) {
 401             let mut path = PathBuf::from(RUN_DIR);
 402             path.push(&mapping);
 403             return unmap_from_backing(&path, loopdev.as_deref());
 404         }
 405     }
 406     Err(format_err!("no mapping for name '{}' found", name.as_ref()))
 407 }
 408
 409 fn minimal_stat(size: i64) -> libc::stat {
 410     let mut stat: libc::stat = unsafe { std::mem::zeroed() };
 411     stat.st_mode = libc::S_IFREG;
 412     stat.st_ino = 1;
 413     stat.st_nlink = 1;
 414     stat.st_size = size;
 415     stat
 416 }