[pbs-devel] [PATCH v2 proxmox-backup 15/20] file-restore: add basic VM/block device support

[Stefan Reiter <s.reiter@proxmox.com>]

Includes methods to start, stop and list QEMU file-restore VMs, as well
as CLI commands do the latter two (start is implicit).

The implementation is abstracted behind the concept of a
"BlockRestoreDriver", so other methods can be implemented later (e.g.
mapping directly to loop devices on the host, using other hypervisors
then QEMU, etc...).

Starting VMs is currently unused but will be needed for further changes.

The design for the QEMU driver uses a locked 'map' file
(/run/user/$UID/restore-vm-map.json) containing a JSON encoding of
currently running VMs. VMs are addressed by a 'name', which is a
systemd-unit encoded combination of repository and snapshot string, thus
uniquely identifying it.

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---

v2:
* this now works per-user, utilizing the setuid helper binary to call QEMU

 src/bin/proxmox-file-restore.rs               |  16 +-
 src/bin/proxmox_client_tools/mod.rs           |  17 +
 src/bin/proxmox_file_restore/block_driver.rs  | 163 +++++++++
 .../proxmox_file_restore/block_driver_qemu.rs | 309 ++++++++++++++++++
 src/bin/proxmox_file_restore/mod.rs           |   5 +
 5 files changed, 507 insertions(+), 3 deletions(-)
 create mode 100644 src/bin/proxmox_file_restore/block_driver.rs
 create mode 100644 src/bin/proxmox_file_restore/block_driver_qemu.rs
 create mode 100644 src/bin/proxmox_file_restore/mod.rs

diff --git a/src/bin/proxmox-file-restore.rs b/src/bin/proxmox-file-restore.rs
index 7957194b..16070925 100644
--- a/src/bin/proxmox-file-restore.rs
+++ b/src/bin/proxmox-file-restore.rs
@@ -35,6 +35,9 @@ use proxmox_client_tools::{
     REPO_URL_SCHEMA,
 };
 
+mod proxmox_file_restore;
+use proxmox_file_restore::*;
+
 enum ExtractPath {
     ListArchives,
     Pxar(String, Vec<u8>),
@@ -51,7 +54,7 @@ fn parse_path(path: String, base64: bool) -> Result<ExtractPath, Error> {
         return Ok(ExtractPath::ListArchives);
     }
 
-    while bytes.len() > 0 && bytes[0] == b'/' {
+    while !bytes.is_empty() && bytes[0] == b'/' {
         bytes.remove(0);
     }
 
@@ -322,7 +325,7 @@ async fn extract(param: Value) -> Result<Value, Error> {
             let file = root
                 .lookup(OsStr::from_bytes(&path))
                 .await?
-                .ok_or(format_err!("error opening '{:?}'", path))?;
+                .ok_or_else(|| format_err!("error opening '{:?}'", path))?;
 
             if let Some(target) = target {
                 extract_sub_dir(target, decoder, OsStr::from_bytes(&path), verbose).await?;
@@ -364,9 +367,16 @@ fn main() {
         .completion_cb("snapshot", complete_group_or_snapshot)
         .completion_cb("target", tools::complete_file_name);
 
+    let status_cmd_def = CliCommand::new(&API_METHOD_STATUS);
+    let stop_cmd_def = CliCommand::new(&API_METHOD_STOP)
+        .arg_param(&["name"])
+        .completion_cb("name", complete_block_driver_ids);
+
     let cmd_def = CliCommandMap::new()
         .insert("list", list_cmd_def)
-        .insert("extract", restore_cmd_def);
+        .insert("extract", restore_cmd_def)
+        .insert("status", status_cmd_def)
+        .insert("stop", stop_cmd_def);
 
     let rpcenv = CliEnvironment::new();
     run_cli_command(
diff --git a/src/bin/proxmox_client_tools/mod.rs b/src/bin/proxmox_client_tools/mod.rs
index 73744ba2..03276993 100644
--- a/src/bin/proxmox_client_tools/mod.rs
+++ b/src/bin/proxmox_client_tools/mod.rs
@@ -372,3 +372,20 @@ pub fn place_xdg_file(
         .and_then(|base| base.place_config_file(file_name).map_err(Error::from))
         .with_context(|| format!("failed to place {} in xdg home", description))
 }
+
+/// Returns a runtime dir owned by the current user
+pub fn get_user_run_dir() -> Result<std::path::PathBuf, Error> {
+    if let Ok(xdg) = base_directories() {
+        if let Ok(path) = xdg.create_runtime_directory("proxmox-backup") {
+            return Ok(path);
+        }
+    }
+    let uid = nix::unistd::Uid::current();
+    let mut path: std::path::PathBuf = format!("/run/user/{}/", uid).into();
+    if !path.exists() {
+        bail!("XDG_RUNTIME_DIR is unavailable, and '{}' doesn't exist", path.display());
+    }
+    path.push("proxmox-backup");
+    std::fs::create_dir_all(&path)?;
+    Ok(path)
+}
diff --git a/src/bin/proxmox_file_restore/block_driver.rs b/src/bin/proxmox_file_restore/block_driver.rs
new file mode 100644
index 00000000..53074f0c
--- /dev/null
+++ b/src/bin/proxmox_file_restore/block_driver.rs
@@ -0,0 +1,163 @@
+//! Abstraction layer over different methods of accessing a block backup
+use anyhow::{bail, Error};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::hash::BuildHasher;
+use std::pin::Pin;
+
+use proxmox_backup::backup::{BackupDir, BackupManifest};
+use proxmox_backup::client::BackupRepository;
+
+use proxmox::api::{api, cli::*};
+
+use super::block_driver_qemu::QemuBlockDriver;
+
+/// Contains details about a snapshot that is to be accessed by block file restore
+pub struct SnapRestoreDetails {
+    pub repo: BackupRepository,
+    pub snapshot: BackupDir,
+    pub manifest: BackupManifest,
+}
+
+/// Return value of a BlockRestoreDriver.status() call, 'id' must be valid for .stop(id)
+pub struct DriverStatus {
+    pub id: String,
+    pub data: Value,
+}
+
+pub type Async<R> = Pin<Box<dyn Future<Output = R> + Send>>;
+
+/// An abstract implementation for retrieving data out of a block file backup
+pub trait BlockRestoreDriver {
+    /// Return status of all running/mapped images, result value is (id, extra data), where id must
+    /// match with the ones returned from list()
+    fn status(&self) -> Async<Result<Vec<DriverStatus>, Error>>;
+    /// Stop/Close a running restore method
+    fn stop(&self, id: String) -> Async<Result<(), Error>>;
+    /// Returned ids must be prefixed with driver type so that they cannot collide between drivers,
+    /// the returned values must be passable to stop()
+    fn list(&self) -> Vec<String>;
+}
+
+#[api()]
+#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Copy)]
+pub enum BlockDriverType {
+    /// Uses a small QEMU/KVM virtual machine to map images securely. Requires PVE-patched QEMU.
+    Qemu,
+}
+
+impl BlockDriverType {
+    fn resolve(&self) -> impl BlockRestoreDriver {
+        match self {
+            BlockDriverType::Qemu => QemuBlockDriver {},
+        }
+    }
+}
+
+const DEFAULT_DRIVER: BlockDriverType = BlockDriverType::Qemu;
+const ALL_DRIVERS: &[BlockDriverType] = &[BlockDriverType::Qemu];
+
+#[api(
+   input: {
+       properties: {
+            "driver": {
+                type: BlockDriverType,
+                optional: true,
+            },
+            "output-format": {
+                schema: OUTPUT_FORMAT,
+                optional: true,
+            },
+        },
+   },
+)]
+/// Retrieve status information about currently running/mapped restore images
+pub async fn status(driver: Option<BlockDriverType>, param: Value) -> Result<(), Error> {
+    let output_format = get_output_format(&param);
+    let text = output_format == "text";
+
+    let mut ret = json!({});
+
+    for dt in ALL_DRIVERS {
+        if driver.is_some() && &driver.unwrap() != dt {
+            continue;
+        }
+
+        let drv_name = format!("{:?}", dt);
+        let drv = dt.resolve();
+        match drv.status().await {
+            Ok(data) if data.is_empty() => {
+                if text {
+                    println!("{}: no mappings", drv_name);
+                } else {
+                    ret[drv_name] = json!({});
+                }
+            }
+            Ok(data) => {
+                if text {
+                    println!("{}:", drv_name);
+                }
+
+                ret[&drv_name]["ids"] = json!([]);
+                for status in data {
+                    if text {
+                        println!("{} \t({})", status.id, status.data);
+                    } else {
+                        ret[&drv_name]["ids"][status.id] = status.data;
+                    }
+                }
+            }
+            Err(err) => {
+                if text {
+                    eprintln!("error getting status from driver '{}' - {}", drv_name, err);
+                } else {
+                    ret[drv_name] = json!({ "error": format!("{}", err) });
+                }
+            }
+        }
+    }
+
+    if !text {
+        format_and_print_result(&ret, &output_format);
+    }
+
+    Ok(())
+}
+
+#[api(
+   input: {
+       properties: {
+            "name": {
+                type: String,
+                description: "The name of the VM to stop.",
+            },
+        },
+   },
+)]
+/// Immediately stop/unmap a given image. Not typically necessary, as VMs will stop themselves
+/// after a timer anyway.
+pub async fn stop(name: String) -> Result<(), Error> {
+    for drv in ALL_DRIVERS.iter().map(BlockDriverType::resolve) {
+        if drv.list().contains(&name) {
+            return drv.stop(name).await;
+        }
+    }
+
+    bail!("no mapping with name '{}' found", name);
+}
+
+/// Autocompletion handler for block mappings
+pub fn complete_block_driver_ids<S: BuildHasher>(
+    _arg: &str,
+    _param: &HashMap<String, String, S>,
+) -> Vec<String> {
+    ALL_DRIVERS
+        .iter()
+        .map(BlockDriverType::resolve)
+        .map(|d| d.list())
+        .flatten()
+        .collect()
+}
diff --git a/src/bin/proxmox_file_restore/block_driver_qemu.rs b/src/bin/proxmox_file_restore/block_driver_qemu.rs
new file mode 100644
index 00000000..5fda5d6f
--- /dev/null
+++ b/src/bin/proxmox_file_restore/block_driver_qemu.rs
@@ -0,0 +1,309 @@
+//! Block file access via a small QEMU restore VM using the PBS block driver in QEMU
+use anyhow::{bail, Error};
+use futures::FutureExt;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use std::collections::HashMap;
+use std::fs::{File, OpenOptions};
+use std::io::{prelude::*, SeekFrom};
+
+use proxmox::tools::fs::lock_file;
+use proxmox_backup::backup::BackupDir;
+use proxmox_backup::buildcfg;
+use proxmox_backup::client::*;
+use proxmox_backup::tools;
+
+use super::block_driver::*;
+use crate::proxmox_client_tools::get_user_run_dir;
+
+const RESTORE_VM_MAP: &str = "restore-vm-map.json";
+
+pub struct QemuBlockDriver {}
+
+#[derive(Clone, Hash, Serialize, Deserialize)]
+struct VMState {
+    pid: i32,
+    cid: i32,
+    ticket: String,
+}
+
+struct VMStateMap {
+    map: HashMap<String, VMState>,
+    file: File,
+}
+
+impl VMStateMap {
+    fn open_file_raw(write: bool) -> Result<File, Error> {
+        let mut path = get_user_run_dir()?;
+        std::fs::create_dir_all(&path)?;
+        path.push(RESTORE_VM_MAP);
+        OpenOptions::new()
+            .read(true)
+            .write(write)
+            .create(write)
+            .open(path)
+            .map_err(Error::from)
+    }
+
+    /// Acquire a lock on the state map and retrieve a deserialized version
+    fn load() -> Result<Self, Error> {
+        let mut file = Self::open_file_raw(true)?;
+        lock_file(&mut file, true, Some(std::time::Duration::from_secs(5)))?;
+        let map = serde_json::from_reader(&file).unwrap_or_default();
+        Ok(Self { map, file })
+    }
+
+    /// Load a read-only copy of the current VM map. Only use for informational purposes, like
+    /// shell auto-completion, for anything requiring consistency use load() !
+    fn load_read_only() -> Result<HashMap<String, VMState>, Error> {
+        let file = Self::open_file_raw(false)?;
+        Ok(serde_json::from_reader(&file).unwrap_or_default())
+    }
+
+    /// Write back a potentially modified state map, consuming the held lock
+    fn write(mut self) -> Result<(), Error> {
+        self.file.seek(SeekFrom::Start(0))?;
+        self.file.set_len(0)?;
+        serde_json::to_writer(self.file, &self.map)?;
+
+        // drop ourselves including file lock
+        Ok(())
+    }
+
+    /// Return the map, but drop the lock immediately
+    fn read_only(self) -> HashMap<String, VMState> {
+        self.map
+    }
+}
+
+fn make_name(repo: &BackupRepository, snap: &BackupDir) -> String {
+    let full = format!("qemu_{}/{}", repo, snap);
+    tools::systemd::escape_unit(&full, false)
+}
+
+/// remove non-responsive VMs from given map, returns 'true' if map was modified
+async fn cleanup_map(map: &mut HashMap<String, VMState>) -> bool {
+    let mut to_remove = Vec::new();
+    for (name, state) in map.iter() {
+        let client = VsockClient::new(state.cid, DEFAULT_VSOCK_PORT, Some(state.ticket.clone()));
+        let res = client
+            .get("api2/json/status", Some(json!({"keep-timeout": true})))
+            .await;
+        if res.is_err() {
+            // VM is not reachable, remove from map and inform user
+            to_remove.push(name.clone());
+            println!(
+                "VM '{}' (pid: {}, cid: {}) was not reachable, removing from map",
+                name, state.pid, state.cid
+            );
+        }
+    }
+
+    for tr in &to_remove {
+        map.remove(tr);
+    }
+
+    !to_remove.is_empty()
+}
+
+fn new_ticket() -> String {
+    proxmox::tools::Uuid::generate().to_string()
+}
+
+async fn ensure_running(details: &SnapRestoreDetails) -> Result<VsockClient, Error> {
+    let name = make_name(&details.repo, &details.snapshot);
+    let mut state = VMStateMap::load()?;
+
+    cleanup_map(&mut state.map).await;
+
+    let new_cid;
+    let vms = match state.map.get(&name) {
+        Some(vm) => {
+            let client = VsockClient::new(vm.cid, DEFAULT_VSOCK_PORT, Some(vm.ticket.clone()));
+            let res = client.get("api2/json/status", None).await;
+            match res {
+                Ok(_) => {
+                    // VM is running and we just reset its timeout, nothing to do
+                    return Ok(client);
+                }
+                Err(err) => {
+                    println!("stale VM detected, restarting ({})", err);
+                    // VM is dead, restart
+                    let vms = start_vm(vm.cid, details).await?;
+                    new_cid = vms.cid;
+                    state.map.insert(name, vms.clone());
+                    vms
+                }
+            }
+        }
+        None => {
+            let mut cid = state
+                .map
+                .iter()
+                .map(|v| v.1.cid)
+                .max()
+                .unwrap_or(0)
+                .wrapping_add(1);
+
+            // offset cid by user id, to avoid unneccessary retries
+            let running_uid = nix::unistd::Uid::current();
+            cid = cid.wrapping_add(running_uid.as_raw() as i32);
+
+            // some low CIDs have special meaning, start at 10 to avoid them
+            cid = cid.max(10);
+
+            let vms = start_vm(cid, details).await?;
+            new_cid = vms.cid;
+            state.map.insert(name, vms.clone());
+            vms
+        }
+    };
+
+    state.write()?;
+    Ok(VsockClient::new(
+        new_cid,
+        DEFAULT_VSOCK_PORT,
+        Some(vms.ticket.clone()),
+    ))
+}
+
+async fn start_vm(cid: i32, details: &SnapRestoreDetails) -> Result<VMState, Error> {
+    let ticket = new_ticket();
+    let mut cmd = std::process::Command::new(buildcfg::PROXMOX_RESTORE_QEMU_HELPER_FN);
+    cmd.arg("start");
+    cmd.arg(details.repo.to_string());
+    cmd.arg(details.snapshot.to_string());
+    cmd.arg(&ticket);
+    cmd.arg(cid.to_string());
+
+    for file in details.manifest.files() {
+        if !file.filename.ends_with(".img.fidx") {
+            continue;
+        }
+        cmd.arg("--files");
+        cmd.arg(&file.filename);
+    }
+
+    // allow the setuid binary to print error messages
+    cmd.stderr(std::process::Stdio::inherit());
+    cmd.stdout(std::process::Stdio::piped());
+
+    let res = tokio::task::block_in_place(|| cmd.spawn()?.wait_with_output())?;
+
+    if res.status.success() {
+        let out = String::from_utf8_lossy(&res.stdout);
+        let val: Value = serde_json::from_str(&out)?;
+        let cid = if let Some(cid) = val["cid"].as_i64() {
+            cid as i32
+        } else {
+            bail!("invalid return from proxmox-restore-qemu-helper: no cid")
+        };
+        let pid = if let Some(pid) = val["pid"].as_i64() {
+            pid as i32
+        } else {
+            bail!("invalid return from proxmox-restore-qemu-helper: no pid")
+        };
+        Ok(VMState {
+            cid,
+            pid,
+            ticket,
+        })
+    } else {
+        bail!("starting VM failed");
+    }
+}
+
+impl BlockRestoreDriver for QemuBlockDriver {
+    fn status(&self) -> Async<Result<Vec<DriverStatus>, Error>> {
+        async move {
+            let mut state_map = VMStateMap::load()?;
+            let modified = cleanup_map(&mut state_map.map).await;
+            let map = if modified {
+                let m = state_map.map.clone();
+                state_map.write()?;
+                m
+            } else {
+                state_map.read_only()
+            };
+            let mut result = Vec::new();
+
+            for (n, s) in map.iter() {
+                let client = VsockClient::new(s.cid, DEFAULT_VSOCK_PORT, Some(s.ticket.clone()));
+                let resp = client
+                    .get("api2/json/status", Some(json!({"keep-timeout": true})))
+                    .await;
+                let name = tools::systemd::unescape_unit(n)
+                    .unwrap_or_else(|_| "<invalid name>".to_owned());
+                let mut extra = json!({"pid": s.pid, "cid": s.cid});
+
+                match resp {
+                    Ok(status) => match status["data"].as_object() {
+                        Some(map) => {
+                            for (k, v) in map.iter() {
+                                extra[k] = v.clone();
+                            }
+                        }
+                        None => {
+                            let err = format!(
+                                "invalid JSON received from /status call: {}",
+                                status.to_string()
+                            );
+                            extra["error"] = json!(err);
+                        }
+                    },
+                    Err(err) => {
+                        let err = format!("error during /status API call: {}", err);
+                        extra["error"] = json!(err);
+                    }
+                }
+
+                result.push(DriverStatus {
+                    id: name,
+                    data: extra,
+                });
+            }
+
+            Ok(result)
+        }
+        .boxed()
+    }
+
+    fn stop(&self, id: String) -> Async<Result<(), Error>> {
+        async move {
+            let name = tools::systemd::escape_unit(&id, false);
+            let mut map = VMStateMap::load()?;
+            let map_mod = cleanup_map(&mut map.map).await;
+            match map.map.get(&name) {
+                Some(state) => {
+                    let client =
+                        VsockClient::new(state.cid, DEFAULT_VSOCK_PORT, Some(state.ticket.clone()));
+                    // ignore errors, this either fails because:
+                    // * the VM is unreachable/dead, in which case we don't want it in the map
+                    // * the call was successful and the connection reset when the VM stopped
+                    let _ = client.get("api2/json/stop", None).await;
+                    map.map.remove(&name);
+                    map.write()?;
+                }
+                None => {
+                    if map_mod {
+                        map.write()?;
+                    }
+                    bail!("VM with name '{}' not found", name);
+                }
+            }
+            Ok(())
+        }
+        .boxed()
+    }
+
+    fn list(&self) -> Vec<String> {
+        match VMStateMap::load_read_only() {
+            Ok(state) => state
+                .iter()
+                .filter_map(|(name, _)| tools::systemd::unescape_unit(&name).ok())
+                .collect(),
+            Err(_) => Vec::new(),
+        }
+    }
+}
diff --git a/src/bin/proxmox_file_restore/mod.rs b/src/bin/proxmox_file_restore/mod.rs
new file mode 100644
index 00000000..52a1259e
--- /dev/null
+++ b/src/bin/proxmox_file_restore/mod.rs
@@ -0,0 +1,5 @@
+//! Block device drivers and tools for single file restore
+pub mod block_driver;
+pub use block_driver::*;
+
+mod block_driver_qemu;
-- 
2.20.1