]> git.proxmox.com Git - proxmox-backup.git/blobdiff - src/bin/proxmox-backup-client.rs
client: refactor catalog upload spawning
[proxmox-backup.git] / src / bin / proxmox-backup-client.rs
index 4a206309d1633bc70fe4c30d1f9a11e3b85c81de..dce8f0b8571a09f5b09ab6cece539fadc1280aaa 100644 (file)
@@ -12,6 +12,7 @@ use futures::future::FutureExt;
 use futures::stream::{StreamExt, TryStreamExt};
 use serde_json::{json, Value};
 use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
 use xdg::BaseDirectories;
 
 use pathpatterns::{MatchEntry, MatchType, PatternFlag};
@@ -32,6 +33,7 @@ use proxmox::{
 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
 
 use proxmox_backup::tools;
+use proxmox_backup::api2::access::user::UserWithTokens;
 use proxmox_backup::api2::types::*;
 use proxmox_backup::api2::version;
 use proxmox_backup::client::*;
@@ -39,6 +41,7 @@ use proxmox_backup::pxar::catalog::*;
 use proxmox_backup::backup::{
     archive_type,
     decrypt_key,
+    rsa_encrypt_key_config,
     verify_chunk_size,
     ArchiveType,
     AsyncReadChunk,
@@ -52,10 +55,11 @@ use proxmox_backup::backup::{
     ChunkStream,
     CryptConfig,
     CryptMode,
-    DataBlob,
     DynamicIndexReader,
+    ENCRYPTED_KEY_BLOB_NAME,
     FixedChunkStream,
     FixedIndexReader,
+    KeyConfig,
     IndexFile,
     MANIFEST_BLOB_NAME,
     Shell,
@@ -192,8 +196,12 @@ pub fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<
     result
 }
 
-fn connect(server: &str, port: u16, userid: &Userid) -> Result<HttpClient, Error> {
+fn connect(repo: &BackupRepository) -> Result<HttpClient, Error> {
+    connect_do(repo.host(), repo.port(), repo.auth_id())
+        .map_err(|err| format_err!("error building client for repository {} - {}", repo, err))
+}
 
+fn connect_do(server: &str, port: u16, auth_id: &Authid) -> Result<HttpClient, Error> {
     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
 
     use std::env::VarError::*;
@@ -211,7 +219,7 @@ fn connect(server: &str, port: u16, userid: &Userid) -> Result<HttpClient, Error
         .fingerprint_cache(true)
         .ticket_cache(true);
 
-    HttpClient::new(server, port, userid, options)
+    HttpClient::new(server, port, auth_id, options)
 }
 
 async fn view_task_result(
@@ -297,9 +305,9 @@ async fn backup_directory<P: AsRef<Path>>(
     )?;
     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 
-    let (mut tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
+    let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 
-    let stream = rx
+    let stream = ReceiverStream::new(rx)
         .map_err(Error::from);
 
     // spawn chunker inside a separate task so that it can run parallel
@@ -365,7 +373,7 @@ async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 
     let repo = extract_repository_from_value(&param)?;
 
-    let client = connect(repo.host(), repo.port(), repo.user())?;
+    let client = connect(&repo)?;
 
     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 
@@ -405,9 +413,9 @@ async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 
     let mut data: Value = result["data"].take();
 
-    let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_GROUPS;
+    let return_type = &proxmox_backup::api2::admin::datastore::API_METHOD_LIST_GROUPS.returns;
 
-    format_and_print_result_full(&mut data, info, &output_format, &options);
+    format_and_print_result_full(&mut data, return_type, &output_format, &options);
 
     Ok(Value::Null)
 }
@@ -422,100 +430,33 @@ async fn list_backup_groups(param: Value) -> Result<Value, Error> {
             group: {
                 type: String,
                 description: "Backup group.",
-                optional: true,
             },
-            "output-format": {
-                schema: OUTPUT_FORMAT,
-                optional: true,
+            "new-owner": {
+                type: Authid,
             },
         }
    }
 )]
-/// List backup snapshots.
-async fn list_snapshots(param: Value) -> Result<Value, Error> {
+/// Change owner of a backup group
+async fn change_backup_owner(group: String, mut param: Value) -> Result<(), Error> {
 
     let repo = extract_repository_from_value(&param)?;
 
-    let output_format = get_output_format(&param);
-
-    let client = connect(repo.host(), repo.port(), repo.user())?;
-
-    let group: Option<BackupGroup> = if let Some(path) = param["group"].as_str() {
-        Some(path.parse()?)
-    } else {
-        None
-    };
-
-    let mut data = api_datastore_list_snapshots(&client, repo.store(), group).await?;
-
-    record_repository(&repo);
-
-    let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
-        let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
-        let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time)?;
-        Ok(snapshot.relative_path().to_str().unwrap().to_owned())
-    };
-
-    let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
-        let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
-        let mut filenames = Vec::new();
-        for file in &item.files {
-            filenames.push(file.filename.to_string());
-        }
-        Ok(tools::format::render_backup_file_list(&filenames[..]))
-    };
-
-    let options = default_table_format_options()
-        .sortby("backup-type", false)
-        .sortby("backup-id", false)
-        .sortby("backup-time", false)
-        .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
-        .column(ColumnConfig::new("size"))
-        .column(ColumnConfig::new("files").renderer(render_files))
-        ;
-
-    let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOTS;
-
-    format_and_print_result_full(&mut data, info, &output_format, &options);
-
-    Ok(Value::Null)
-}
-
-#[api(
-   input: {
-        properties: {
-            repository: {
-                schema: REPO_URL_SCHEMA,
-                optional: true,
-            },
-            snapshot: {
-                type: String,
-                description: "Snapshot path.",
-             },
-        }
-   }
-)]
-/// Forget (remove) backup snapshots.
-async fn forget_snapshots(param: Value) -> Result<Value, Error> {
-
-    let repo = extract_repository_from_value(&param)?;
+    let mut client = connect(&repo)?;
 
-    let path = tools::required_string_param(&param, "snapshot")?;
-    let snapshot: BackupDir = path.parse()?;
+    param.as_object_mut().unwrap().remove("repository");
 
-    let mut client = connect(repo.host(), repo.port(), repo.user())?;
+    let group: BackupGroup = group.parse()?;
 
-    let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
+    param["backup-type"] = group.backup_type().into();
+    param["backup-id"] = group.backup_id().into();
 
-    let result = client.delete(&path, Some(json!({
-        "backup-type": snapshot.group().backup_type(),
-        "backup-id": snapshot.group().backup_id(),
-        "backup-time": snapshot.backup_time(),
-    }))).await?;
+    let path = format!("api2/json/admin/datastore/{}/change-owner", repo.store());
+    client.post(&path, Some(param)).await?;
 
     record_repository(&repo);
 
-    Ok(result)
+    Ok(())
 }
 
 #[api(
@@ -533,7 +474,7 @@ async fn api_login(param: Value) -> Result<Value, Error> {
 
     let repo = extract_repository_from_value(&param)?;
 
-    let client = connect(repo.host(), repo.port(), repo.user())?;
+    let client = connect(&repo)?;
     client.login().await?;
 
     record_repository(&repo);
@@ -590,7 +531,7 @@ async fn api_version(param: Value) -> Result<(), Error> {
 
     let repo = extract_repository_from_value(&param);
     if let Ok(repo) = repo {
-        let client = connect(repo.host(), repo.port(), repo.user())?;
+        let client = connect(&repo)?;
 
         match client.get("api2/json/version", None).await {
             Ok(mut result) => version_info["server"] = result["data"].take(),
@@ -611,58 +552,6 @@ async fn api_version(param: Value) -> Result<(), Error> {
     Ok(())
 }
 
-
-#[api(
-   input: {
-        properties: {
-            repository: {
-                schema: REPO_URL_SCHEMA,
-                optional: true,
-            },
-            snapshot: {
-                type: String,
-                description: "Snapshot path.",
-             },
-            "output-format": {
-                schema: OUTPUT_FORMAT,
-                optional: true,
-            },
-        }
-   }
-)]
-/// List snapshot files.
-async fn list_snapshot_files(param: Value) -> Result<Value, Error> {
-
-    let repo = extract_repository_from_value(&param)?;
-
-    let path = tools::required_string_param(&param, "snapshot")?;
-    let snapshot: BackupDir = path.parse()?;
-
-    let output_format = get_output_format(&param);
-
-    let client = connect(repo.host(), repo.port(), repo.user())?;
-
-    let path = format!("api2/json/admin/datastore/{}/files", repo.store());
-
-    let mut result = client.get(&path, Some(json!({
-        "backup-type": snapshot.group().backup_type(),
-        "backup-id": snapshot.group().backup_id(),
-        "backup-time": snapshot.backup_time(),
-    }))).await?;
-
-    record_repository(&repo);
-
-    let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOT_FILES;
-
-    let mut data: Value = result["data"].take();
-
-    let options = default_table_format_options();
-
-    format_and_print_result_full(&mut data, info, &output_format, &options);
-
-    Ok(Value::Null)
-}
-
 #[api(
     input: {
         properties: {
@@ -684,7 +573,7 @@ async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 
     let output_format = get_output_format(&param);
 
-    let mut client = connect(repo.host(), repo.port(), repo.user())?;
+    let mut client = connect(&repo)?;
 
     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 
@@ -697,21 +586,21 @@ async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
     Ok(Value::Null)
 }
 
+struct CatalogUploadResult {
+    catalog_writer: Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
+    result: tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>,
+}
+
 fn spawn_catalog_upload(
     client: Arc<BackupWriter>,
     encrypt: bool,
-) -> Result<
-        (
-            Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
-            tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>
-        ), Error>
-{
+) -> Result<CatalogUploadResult, Error> {
     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
     let catalog_stream = crate::tools::StdChannelStream(catalog_rx);
     let catalog_chunk_size = 512*1024;
     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 
-    let catalog = Arc::new(Mutex::new(CatalogWriter::new(crate::tools::StdChannelWriter::new(catalog_tx))?));
+    let catalog_writer = Arc::new(Mutex::new(CatalogWriter::new(crate::tools::StdChannelWriter::new(catalog_tx))?));
 
     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 
@@ -728,7 +617,7 @@ fn spawn_catalog_upload(
         let _ = catalog_result_tx.send(catalog_upload_result);
     });
 
-    Ok((catalog, catalog_result_rx))
+    Ok(CatalogUploadResult { catalog_writer, result: catalog_result_rx })
 }
 
 fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Error> {
@@ -758,7 +647,10 @@ fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Err
     let keydata = match (keyfile, key_fd) {
         (None, None) => None,
         (Some(_), Some(_)) => bail!("--keyfile and --keyfd are mutually exclusive"),
-        (Some(keyfile), None) => Some(file_get_contents(keyfile)?),
+        (Some(keyfile), None) => {
+            eprintln!("Using encryption key file: {}", keyfile);
+            Some(file_get_contents(keyfile)?)
+        },
         (None, Some(fd)) => {
             let input = unsafe { std::fs::File::from_raw_fd(fd) };
             let mut data = Vec::new();
@@ -766,6 +658,7 @@ fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Err
                 .map_err(|err| {
                     format_err!("error reading encryption key from fd {}: {}", fd, err)
                 })?;
+            eprintln!("Using encryption key from file descriptor");
             Some(data)
         }
     };
@@ -773,7 +666,10 @@ fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Err
     Ok(match (keydata, crypt_mode) {
         // no parameters:
         (None, None) => match key::read_optional_default_encryption_key()? {
-            Some(key) => (Some(key), CryptMode::Encrypt),
+            Some(key) => {
+                eprintln!("Encrypting with default encryption key!");
+                (Some(key), CryptMode::Encrypt)
+            },
             None => (None, CryptMode::None),
         },
 
@@ -783,7 +679,10 @@ fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Err
         // just --crypt-mode other than none
         (None, Some(crypt_mode)) => match key::read_optional_default_encryption_key()? {
             None => bail!("--crypt-mode without --keyfile and no default key file available"),
-            Some(key) => (Some(key), crypt_mode),
+            Some(key) => {
+                eprintln!("Encrypting with default encryption key!");
+                (Some(key), crypt_mode)
+            },
         }
 
         // just --keyfile
@@ -821,6 +720,11 @@ fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Err
                    description: "Path to file.",
                }
            },
+           "all-file-systems": {
+               type: Boolean,
+               description: "Include all mounted subdirectories.",
+               optional: true,
+           },
            keyfile: {
                schema: KEYFILE_SCHEMA,
                optional: true,
@@ -994,9 +898,9 @@ async fn create_backup(
         }
     }
 
-    let backup_time = backup_time_opt.unwrap_or_else(|| epoch_i64());
+    let backup_time = backup_time_opt.unwrap_or_else(epoch_i64);
 
-    let client = connect(repo.host(), repo.port(), repo.user())?;
+    let client = connect(&repo)?;
     record_repository(&repo);
 
     println!("Starting backup: {}/{}/{}", backup_type, backup_id, BackupDir::backup_time_to_string(backup_time)?);
@@ -1010,7 +914,8 @@ async fn create_backup(
     let (crypt_config, rsa_encrypted_key) = match keydata {
         None => (None, None),
         Some(key) => {
-            let (key, created) = decrypt_key(&key, &key::get_encryption_key_password)?;
+            let (key, created, fingerprint) = decrypt_key(&key, &key::get_encryption_key_password)?;
+            println!("Encryption key fingerprint: {}", fingerprint);
 
             let crypt_config = CryptConfig::new(key)?;
 
@@ -1018,7 +923,13 @@ async fn create_backup(
                 Some(ref path) if path.exists() => {
                     let pem_data = file_get_contents(path)?;
                     let rsa = openssl::rsa::Rsa::public_key_from_pem(&pem_data)?;
-                    let enc_key = crypt_config.generate_rsa_encoded_key(rsa, created)?;
+
+                    let mut key_config = KeyConfig::without_password(key)?;
+                    key_config.created = created; // keep original value
+
+                    let enc_key = rsa_encrypt_key_config(rsa, &key_config)?;
+                    println!("Master key '{:?}'", path);
+
                     (Some(Arc::new(crypt_config)), Some(enc_key))
                 }
                 _ => (Some(Arc::new(crypt_config)), None),
@@ -1037,8 +948,40 @@ async fn create_backup(
         false
     ).await?;
 
-    let previous_manifest = if let Ok(previous_manifest) = client.download_previous_manifest().await {
-        Some(Arc::new(previous_manifest))
+    let download_previous_manifest = match client.previous_backup_time().await {
+        Ok(Some(backup_time)) => {
+            println!(
+                "Downloading previous manifest ({})",
+                strftime_local("%c", backup_time)?
+            );
+            true
+        }
+        Ok(None) => {
+            println!("No previous manifest available.");
+            false
+        }
+        Err(_) => {
+            // Fallback for outdated server, TODO remove/bubble up with 2.0
+            true
+        }
+    };
+
+    let previous_manifest = if download_previous_manifest {
+        match client.download_previous_manifest().await {
+            Ok(previous_manifest) => {
+                match previous_manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref)) {
+                    Ok(()) => Some(Arc::new(previous_manifest)),
+                    Err(err) => {
+                        println!("Couldn't re-use previous manifest - {}", err);
+                        None
+                    }
+                }
+            }
+            Err(err) => {
+                println!("Couldn't download previous manifest - {}", err);
+                None
+            }
+        }
     } else {
         None
     };
@@ -1047,7 +990,7 @@ async fn create_backup(
     let mut manifest = BackupManifest::new(snapshot);
 
     let mut catalog = None;
-    let mut catalog_result_tx = None;
+    let mut catalog_result_rx = None;
 
     for (backup_type, filename, target, size) in upload_list {
         match backup_type {
@@ -1068,9 +1011,9 @@ async fn create_backup(
             BackupSpecificationType::PXAR => {
                 // start catalog upload on first use
                 if catalog.is_none() {
-                    let (cat, res) = spawn_catalog_upload(client.clone(), crypt_mode == CryptMode::Encrypt)?;
-                    catalog = Some(cat);
-                    catalog_result_tx = Some(res);
+                    let catalog_upload_res = spawn_catalog_upload(client.clone(), crypt_mode == CryptMode::Encrypt)?;
+                    catalog = Some(catalog_upload_res.catalog_writer);
+                    catalog_result_rx = Some(catalog_upload_res.result);
                 }
                 let catalog = catalog.as_ref().unwrap();
 
@@ -1122,28 +1065,20 @@ async fn create_backup(
 
         drop(catalog); // close upload stream
 
-        if let Some(catalog_result_rx) = catalog_result_tx {
+        if let Some(catalog_result_rx) = catalog_result_rx {
             let stats = catalog_result_rx.await??;
             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypt_mode)?;
         }
     }
 
     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
-        let target = "rsa-encrypted.key.blob";
+        let target = ENCRYPTED_KEY_BLOB_NAME;
         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
         let stats = client
             .upload_blob_from_data(rsa_encrypted_key, target, false, false)
             .await?;
         manifest.add_file(target.to_string(), stats.size, stats.csum, crypt_mode)?;
 
-        // openssl rsautl -decrypt -inkey master-private.pem -in rsa-encrypted.key -out t
-        /*
-        let mut buffer2 = vec![0u8; rsa.size() as usize];
-        let pem_data = file_get_contents("master-private.pem")?;
-        let rsa = openssl::rsa::Rsa::private_key_from_pem(&pem_data)?;
-        let len = rsa.private_decrypt(&buffer, &mut buffer2, openssl::rsa::Padding::PKCS1)?;
-        println!("TEST {} {:?}", len, buffer2);
-         */
     }
     // create manifest (index.json)
     // manifests are never encrypted, but include a signature
@@ -1299,7 +1234,7 @@ async fn restore(param: Value) -> Result<Value, Error> {
 
     let archive_name = tools::required_string_param(&param, "archive-name")?;
 
-    let client = connect(repo.host(), repo.port(), repo.user())?;
+    let client = connect(&repo)?;
 
     record_repository(&repo);
 
@@ -1321,7 +1256,8 @@ async fn restore(param: Value) -> Result<Value, Error> {
     let crypt_config = match keydata {
         None => None,
         Some(key) => {
-            let (key, _) = decrypt_key(&key, &key::get_encryption_key_password)?;
+            let (key, _, fingerprint) = decrypt_key(&key, &key::get_encryption_key_password)?;
+            eprintln!("Encryption key fingerprint: '{}'", fingerprint);
             Some(Arc::new(CryptConfig::new(key)?))
         }
     };
@@ -1336,9 +1272,15 @@ async fn restore(param: Value) -> Result<Value, Error> {
         true,
     ).await?;
 
+    let (archive_name, archive_type) = parse_archive_type(archive_name);
+
     let (manifest, backup_index_data) = client.download_manifest().await?;
 
-    let (archive_name, archive_type) = parse_archive_type(archive_name);
+    if archive_name == ENCRYPTED_KEY_BLOB_NAME && crypt_config.is_none() {
+        eprintln!("Restoring encrypted key blob without original key - skipping manifest fingerprint check!")
+    } else {
+        manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref))?;
+    }
 
     if archive_name == MANIFEST_BLOB_NAME {
         if let Some(target) = target {
@@ -1433,81 +1375,6 @@ async fn restore(param: Value) -> Result<Value, Error> {
     Ok(Value::Null)
 }
 
-#[api(
-   input: {
-       properties: {
-           repository: {
-               schema: REPO_URL_SCHEMA,
-               optional: true,
-           },
-           snapshot: {
-               type: String,
-               description: "Group/Snapshot path.",
-           },
-           logfile: {
-               type: String,
-               description: "The path to the log file you want to upload.",
-           },
-           keyfile: {
-               schema: KEYFILE_SCHEMA,
-               optional: true,
-           },
-           "keyfd": {
-               schema: KEYFD_SCHEMA,
-               optional: true,
-           },
-           "crypt-mode": {
-               type: CryptMode,
-               optional: true,
-           },
-       }
-   }
-)]
-/// Upload backup log file.
-async fn upload_log(param: Value) -> Result<Value, Error> {
-
-    let logfile = tools::required_string_param(&param, "logfile")?;
-    let repo = extract_repository_from_value(&param)?;
-
-    let snapshot = tools::required_string_param(&param, "snapshot")?;
-    let snapshot: BackupDir = snapshot.parse()?;
-
-    let mut client = connect(repo.host(), repo.port(), repo.user())?;
-
-    let (keydata, crypt_mode) = keyfile_parameters(&param)?;
-
-    let crypt_config = match keydata {
-        None => None,
-        Some(key) => {
-            let (key, _created) = decrypt_key(&key, &key::get_encryption_key_password)?;
-            let crypt_config = CryptConfig::new(key)?;
-            Some(Arc::new(crypt_config))
-        }
-    };
-
-    let data = file_get_contents(logfile)?;
-
-    // fixme: howto sign log?
-    let blob = match crypt_mode {
-        CryptMode::None | CryptMode::SignOnly => DataBlob::encode(&data, None, true)?,
-        CryptMode::Encrypt => DataBlob::encode(&data, crypt_config.as_ref().map(Arc::as_ref), true)?,
-    };
-
-    let raw_data = blob.into_inner();
-
-    let path = format!("api2/json/admin/datastore/{}/upload-backup-log", repo.store());
-
-    let args = json!({
-        "backup-type": snapshot.group().backup_type(),
-        "backup-id":  snapshot.group().backup_id(),
-        "backup-time": snapshot.backup_time(),
-    });
-
-    let body = hyper::Body::from(raw_data);
-
-    client.upload("application/octet-stream", body, &path, Some(args)).await
-}
-
 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
     &ApiHandler::Async(&prune),
     &ObjectSchema::new(
@@ -1543,7 +1410,7 @@ fn prune<'a>(
 async fn prune_async(mut param: Value) -> Result<Value, Error> {
     let repo = extract_repository_from_value(&param)?;
 
-    let mut client = connect(repo.host(), repo.port(), repo.user())?;
+    let mut client = connect(&repo)?;
 
     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
 
@@ -1589,18 +1456,18 @@ async fn prune_async(mut param: Value) -> Result<Value, Error> {
         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
         ;
 
-    let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_PRUNE;
+    let return_type = &proxmox_backup::api2::admin::datastore::API_METHOD_PRUNE.returns;
 
     let mut data = result["data"].take();
 
     if quiet {
         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
             item["keep"].as_bool() == Some(false)
-        }).map(|v| v.clone()).collect();
+        }).cloned().collect();
         data = list.into();
     }
 
-    format_and_print_result_full(&mut data, info, &output_format, &options);
+    format_and_print_result_full(&mut data, return_type, &output_format, &options);
 
     Ok(Value::Null)
 }
@@ -1617,7 +1484,10 @@ async fn prune_async(mut param: Value) -> Result<Value, Error> {
                optional: true,
            },
        }
-   }
+   },
+    returns: {
+        type: StorageStatus,
+    },
 )]
 /// Get repository status.
 async fn status(param: Value) -> Result<Value, Error> {
@@ -1626,7 +1496,7 @@ async fn status(param: Value) -> Result<Value, Error> {
 
     let output_format = get_output_format(&param);
 
-    let client = connect(repo.host(), repo.port(), repo.user())?;
+    let client = connect(&repo)?;
 
     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
 
@@ -1650,9 +1520,9 @@ async fn status(param: Value) -> Result<Value, Error> {
         .column(ColumnConfig::new("used").renderer(render_total_percentage))
         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
 
-    let schema = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_STATUS;
+    let return_type = &API_METHOD_STATUS.returns;
 
-    format_and_print_result_full(&mut data, schema, &output_format, &options);
+    format_and_print_result_full(&mut data, return_type, &output_format, &options);
 
     Ok(Value::Null)
 }
@@ -1671,7 +1541,7 @@ async fn try_get(repo: &BackupRepository, url: &str) -> Value {
         .fingerprint_cache(true)
         .ticket_cache(true);
 
-    let client = match HttpClient::new(repo.host(), repo.port(), repo.user(), options) {
+    let client = match HttpClient::new(repo.host(), repo.port(), repo.auth_id(), options) {
         Ok(v) => v,
         _ => return Value::Null,
     };
@@ -1817,17 +1687,29 @@ async fn complete_server_file_name_do(param: &HashMap<String, String>) -> Vec<St
 fn complete_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
     complete_server_file_name(arg, param)
         .iter()
-        .map(|v| tools::format::strip_server_file_expenstion(&v))
+        .map(|v| tools::format::strip_server_file_extension(&v))
         .collect()
 }
 
 pub fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
     complete_server_file_name(arg, param)
         .iter()
-        .filter_map(|v| {
-            let name = tools::format::strip_server_file_expenstion(&v);
-            if name.ends_with(".pxar") {
-                Some(name)
+        .filter_map(|name| {
+            if name.ends_with(".pxar.didx") {
+                Some(tools::format::strip_server_file_extension(name))
+            } else {
+                None
+            }
+        })
+        .collect()
+}
+
+pub fn complete_img_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+    complete_server_file_name(arg, param)
+        .iter()
+        .filter_map(|name| {
+            if name.ends_with(".img.fidx") {
+                Some(tools::format::strip_server_file_extension(name))
             } else {
                 None
             }
@@ -1849,6 +1731,33 @@ fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<Stri
     result
 }
 
+fn complete_auth_id(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+    proxmox_backup::tools::runtime::main(async { complete_auth_id_do(param).await })
+}
+
+async fn complete_auth_id_do(param: &HashMap<String, String>) -> Vec<String> {
+
+    let mut result = vec![];
+
+    let repo = match extract_repository_from_map(param) {
+        Some(v) => v,
+        _ => return result,
+    };
+
+    let data = try_get(&repo, "api2/json/access/users?include_tokens=true").await;
+
+    if let Ok(parsed) = serde_json::from_value::<Vec<UserWithTokens>>(data) {
+        for user in parsed {
+            result.push(user.userid.to_string());
+            for token in user.tokens {
+                result.push(token.tokenid.to_string());
+            }
+        }
+    };
+
+    result
+}
+
 use proxmox_backup::client::RemoteChunkReader;
 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
 /// async use!
@@ -1903,26 +1812,9 @@ fn main() {
         .completion_cb("repository", complete_repository)
         .completion_cb("keyfile", tools::complete_file_name);
 
-    let upload_log_cmd_def = CliCommand::new(&API_METHOD_UPLOAD_LOG)
-        .arg_param(&["snapshot", "logfile"])
-        .completion_cb("snapshot", complete_backup_snapshot)
-        .completion_cb("logfile", tools::complete_file_name)
-        .completion_cb("keyfile", tools::complete_file_name)
-        .completion_cb("repository", complete_repository);
-
     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
         .completion_cb("repository", complete_repository);
 
-    let snapshots_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOTS)
-        .arg_param(&["group"])
-        .completion_cb("group", complete_backup_group)
-        .completion_cb("repository", complete_repository);
-
-    let forget_cmd_def = CliCommand::new(&API_METHOD_FORGET_SNAPSHOTS)
-        .arg_param(&["snapshot"])
-        .completion_cb("repository", complete_repository)
-        .completion_cb("snapshot", complete_backup_snapshot);
-
     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
         .completion_cb("repository", complete_repository);
 
@@ -1933,11 +1825,6 @@ fn main() {
         .completion_cb("archive-name", complete_archive_name)
         .completion_cb("target", tools::complete_file_name);
 
-    let files_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOT_FILES)
-        .arg_param(&["snapshot"])
-        .completion_cb("repository", complete_repository)
-        .completion_cb("snapshot", complete_backup_snapshot);
-
     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
         .arg_param(&["group"])
         .completion_cb("group", complete_backup_group)
@@ -1955,18 +1842,21 @@ fn main() {
     let version_cmd_def = CliCommand::new(&API_METHOD_API_VERSION)
         .completion_cb("repository", complete_repository);
 
+    let change_owner_cmd_def = CliCommand::new(&API_METHOD_CHANGE_BACKUP_OWNER)
+        .arg_param(&["group", "new-owner"])
+        .completion_cb("group", complete_backup_group)
+        .completion_cb("new-owner",  complete_auth_id)
+        .completion_cb("repository", complete_repository);
+
     let cmd_def = CliCommandMap::new()
         .insert("backup", backup_cmd_def)
-        .insert("upload-log", upload_log_cmd_def)
-        .insert("forget", forget_cmd_def)
         .insert("garbage-collect", garbage_collect_cmd_def)
         .insert("list", list_cmd_def)
         .insert("login", login_cmd_def)
         .insert("logout", logout_cmd_def)
         .insert("prune", prune_cmd_def)
         .insert("restore", restore_cmd_def)
-        .insert("snapshots", snapshots_cmd_def)
-        .insert("files", files_cmd_def)
+        .insert("snapshot", snapshot_mgtm_cli())
         .insert("status", status_cmd_def)
         .insert("key", key::cli())
         .insert("mount", mount_cmd_def())
@@ -1975,7 +1865,14 @@ fn main() {
         .insert("catalog", catalog_mgmt_cli())
         .insert("task", task_mgmt_cli())
         .insert("version", version_cmd_def)
-        .insert("benchmark", benchmark_cmd_def);
+        .insert("benchmark", benchmark_cmd_def)
+        .insert("change-owner", change_owner_cmd_def)
+
+        .alias(&["files"], &["snapshot", "files"])
+        .alias(&["forget"], &["snapshot", "forget"])
+        .alias(&["upload-log"], &["snapshot", "upload-log"])
+        .alias(&["snapshots"], &["snapshot", "list"])
+        ;
 
     let rpcenv = CliEnvironment::new();
     run_cli_command(cmd_def, rpcenv, Some(|future| {