proxmox-backup-client/src/main.rs

   1 use std::collections::HashSet;
   2 use std::io::{self, Read, Seek, SeekFrom, Write};
   3 use std::path::{Path, PathBuf};
   4 use std::pin::Pin;
   5 use std::sync::{Arc, Mutex};
   6 use std::task::Context;
   7
   8 use anyhow::{bail, format_err, Error};
   9 use futures::stream::{StreamExt, TryStreamExt};
  10 use serde::Deserialize;
  11 use serde_json::{json, Value};
  12 use tokio::sync::mpsc;
  13 use tokio_stream::wrappers::ReceiverStream;
  14 use xdg::BaseDirectories;
  15
  16 use pathpatterns::{MatchEntry, MatchType, PatternFlag};
  17 use proxmox_async::blocking::TokioWriterAdapter;
  18 use proxmox_io::StdChannelWriter;
  19 use proxmox_router::{cli::*, ApiMethod, RpcEnvironment};
  20 use proxmox_schema::api;
  21 use proxmox_sys::fs::{file_get_json, image_size, replace_file, CreateOptions};
  22 use proxmox_time::{epoch_i64, strftime_local};
  23 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
  24
  25 use pbs_api_types::{
  26     Authid, BackupDir, BackupGroup, BackupNamespace, BackupPart, BackupType, CryptMode,
  27     Fingerprint, GroupListItem, HumanByte, PruneListItem, PruneOptions, RateLimitConfig,
  28     SnapshotListItem, StorageStatus, BACKUP_ID_SCHEMA, BACKUP_NAMESPACE_SCHEMA, BACKUP_TIME_SCHEMA,
  29     BACKUP_TYPE_SCHEMA, TRAFFIC_CONTROL_BURST_SCHEMA, TRAFFIC_CONTROL_RATE_SCHEMA,
  30 };
  31 use pbs_client::catalog_shell::Shell;
  32 use pbs_client::tools::{
  33     complete_archive_name, complete_auth_id, complete_backup_group, complete_backup_snapshot,
  34     complete_backup_source, complete_chunk_size, complete_group_or_snapshot,
  35     complete_img_archive_name, complete_pxar_archive_name, complete_repository, connect,
  36     connect_rate_limited, extract_repository_from_value,
  37     key_source::{
  38         crypto_parameters, format_key_source, get_encryption_key_password, KEYFD_SCHEMA,
  39         KEYFILE_SCHEMA, MASTER_PUBKEY_FD_SCHEMA, MASTER_PUBKEY_FILE_SCHEMA,
  40     },
  41     CHUNK_SIZE_SCHEMA, REPO_URL_SCHEMA,
  42 };
  43 use pbs_client::{
  44     delete_ticket_info, parse_backup_specification, view_task_result, BackupReader,
  45     BackupRepository, BackupSpecificationType, BackupStats, BackupWriter, ChunkStream,
  46     FixedChunkStream, HttpClient, PxarBackupStream, RemoteChunkReader, UploadOptions,
  47     BACKUP_SOURCE_SCHEMA,
  48 };
  49 use pbs_config::key_config::{decrypt_key, rsa_encrypt_key_config, KeyConfig};
  50 use pbs_datastore::catalog::{BackupCatalogWriter, CatalogReader, CatalogWriter};
  51 use pbs_datastore::chunk_store::verify_chunk_size;
  52 use pbs_datastore::dynamic_index::{BufferedDynamicReader, DynamicIndexReader};
  53 use pbs_datastore::fixed_index::FixedIndexReader;
  54 use pbs_datastore::index::IndexFile;
  55 use pbs_datastore::manifest::{
  56     archive_type, ArchiveType, BackupManifest, ENCRYPTED_KEY_BLOB_NAME, MANIFEST_BLOB_NAME,
  57 };
  58 use pbs_datastore::read_chunk::AsyncReadChunk;
  59 use pbs_datastore::CATALOG_NAME;
  60 use pbs_tools::crypt_config::CryptConfig;
  61 use pbs_tools::json;
  62
  63 mod benchmark;
  64 pub use benchmark::*;
  65 mod mount;
  66 pub use mount::*;
  67 mod task;
  68 pub use task::*;
  69 mod catalog;
  70 pub use catalog::*;
  71 mod snapshot;
  72 pub use snapshot::*;
  73 pub mod key;
  74 pub mod namespace;
  75
  76 fn record_repository(repo: &BackupRepository) {
  77     let base = match BaseDirectories::with_prefix("proxmox-backup") {
  78         Ok(v) => v,
  79         _ => return,
  80     };
  81
  82     // usually $HOME/.cache/proxmox-backup/repo-list
  83     let path = match base.place_cache_file("repo-list") {
  84         Ok(v) => v,
  85         _ => return,
  86     };
  87
  88     let mut data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
  89
  90     let repo = repo.to_string();
  91
  92     data[&repo] = json! { data[&repo].as_i64().unwrap_or(0) + 1 };
  93
  94     let mut map = serde_json::map::Map::new();
  95
  96     loop {
  97         let mut max_used = 0;
  98         let mut max_repo = None;
  99         for (repo, count) in data.as_object().unwrap() {
 100             if map.contains_key(repo) {
 101                 continue;
 102             }
 103             if let Some(count) = count.as_i64() {
 104                 if count > max_used {
 105                     max_used = count;
 106                     max_repo = Some(repo);
 107                 }
 108             }
 109         }
 110         if let Some(repo) = max_repo {
 111             map.insert(repo.to_owned(), json!(max_used));
 112         } else {
 113             break;
 114         }
 115         if map.len() > 10 {
 116             // store max. 10 repos
 117             break;
 118         }
 119     }
 120
 121     let new_data = json!(map);
 122
 123     let _ = replace_file(
 124         path,
 125         new_data.to_string().as_bytes(),
 126         CreateOptions::new(),
 127         false,
 128     );
 129 }
 130
 131 async fn api_datastore_list_snapshots(
 132     client: &HttpClient,
 133     store: &str,
 134     ns: &BackupNamespace,
 135     group: Option<&BackupGroup>,
 136 ) -> Result<Value, Error> {
 137     let path = format!("api2/json/admin/datastore/{}/snapshots", store);
 138
 139     let mut args = match group {
 140         Some(group) => serde_json::to_value(group)?,
 141         None => json!({}),
 142     };
 143     if !ns.is_root() {
 144         args["ns"] = serde_json::to_value(ns)?;
 145     }
 146
 147     let mut result = client.get(&path, Some(args)).await?;
 148
 149     Ok(result["data"].take())
 150 }
 151
 152 pub async fn api_datastore_latest_snapshot(
 153     client: &HttpClient,
 154     store: &str,
 155     ns: &BackupNamespace,
 156     group: BackupGroup,
 157 ) -> Result<BackupDir, Error> {
 158     let list = api_datastore_list_snapshots(client, store, ns, Some(&group)).await?;
 159     let mut list: Vec<SnapshotListItem> = serde_json::from_value(list)?;
 160
 161     if list.is_empty() {
 162         bail!("backup group {} does not contain any snapshots.", group);
 163     }
 164
 165     list.sort_unstable_by(|a, b| b.backup.time.cmp(&a.backup.time));
 166
 167     Ok((group, list[0].backup.time).into())
 168 }
 169
 170 pub async fn dir_or_last_from_group(
 171     client: &HttpClient,
 172     repo: &BackupRepository,
 173     ns: &BackupNamespace,
 174     path: &str,
 175 ) -> Result<BackupDir, Error> {
 176     match path.parse::<BackupPart>()? {
 177         BackupPart::Dir(dir) => Ok(dir),
 178         BackupPart::Group(group) => {
 179             api_datastore_latest_snapshot(&client, repo.store(), ns, group).await
 180         }
 181     }
 182 }
 183
 184 async fn backup_directory<P: AsRef<Path>>(
 185     client: &BackupWriter,
 186     dir_path: P,
 187     archive_name: &str,
 188     chunk_size: Option<usize>,
 189     catalog: Arc<Mutex<CatalogWriter<TokioWriterAdapter<StdChannelWriter<Error>>>>>,
 190     pxar_create_options: pbs_client::pxar::PxarCreateOptions,
 191     upload_options: UploadOptions,
 192 ) -> Result<BackupStats, Error> {
 193     let pxar_stream = PxarBackupStream::open(dir_path.as_ref(), catalog, pxar_create_options)?;
 194     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 195
 196     let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 197
 198     let stream = ReceiverStream::new(rx).map_err(Error::from);
 199
 200     // spawn chunker inside a separate task so that it can run parallel
 201     tokio::spawn(async move {
 202         while let Some(v) = chunk_stream.next().await {
 203             let _ = tx.send(v).await;
 204         }
 205     });
 206
 207     if upload_options.fixed_size.is_some() {
 208         bail!("cannot backup directory with fixed chunk size!");
 209     }
 210
 211     let stats = client
 212         .upload_stream(archive_name, stream, upload_options)
 213         .await?;
 214
 215     Ok(stats)
 216 }
 217
 218 async fn backup_image<P: AsRef<Path>>(
 219     client: &BackupWriter,
 220     image_path: P,
 221     archive_name: &str,
 222     chunk_size: Option<usize>,
 223     upload_options: UploadOptions,
 224 ) -> Result<BackupStats, Error> {
 225     let path = image_path.as_ref().to_owned();
 226
 227     let file = tokio::fs::File::open(path).await?;
 228
 229     let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 230         .map_err(Error::from);
 231
 232     let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4 * 1024 * 1024));
 233
 234     if upload_options.fixed_size.is_none() {
 235         bail!("cannot backup image with dynamic chunk size!");
 236     }
 237
 238     let stats = client
 239         .upload_stream(archive_name, stream, upload_options)
 240         .await?;
 241
 242     Ok(stats)
 243 }
 244
 245 pub fn optional_ns_param(param: &Value) -> Result<BackupNamespace, Error> {
 246     Ok(match param.get("ns") {
 247         Some(Value::String(ns)) => ns.parse()?,
 248         Some(_) => bail!("invalid namespace parameter"),
 249         None => BackupNamespace::root(),
 250     })
 251 }
 252
 253 #[api(
 254    input: {
 255         properties: {
 256             repository: {
 257                 schema: REPO_URL_SCHEMA,
 258                 optional: true,
 259             },
 260             "ns": {
 261                 type: BackupNamespace,
 262                 optional: true,
 263             },
 264             "output-format": {
 265                 schema: OUTPUT_FORMAT,
 266                 optional: true,
 267             },
 268         }
 269    }
 270 )]
 271 /// List backup groups.
 272 async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 273     let output_format = get_output_format(&param);
 274
 275     let repo = extract_repository_from_value(&param)?;
 276
 277     let client = connect(&repo)?;
 278
 279     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 280
 281     let backup_ns = optional_ns_param(&param)?;
 282     let mut result = client
 283         .get(
 284             &path,
 285             match backup_ns.is_root() {
 286                 true => None,
 287                 false => Some(json!({ "ns": backup_ns })),
 288             },
 289         )
 290         .await?;
 291
 292     record_repository(&repo);
 293
 294     let render_group_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 295         let item = GroupListItem::deserialize(record)?;
 296         Ok(item.backup.to_string())
 297     };
 298
 299     let render_last_backup = |_v: &Value, record: &Value| -> Result<String, Error> {
 300         let item = GroupListItem::deserialize(record)?;
 301         let snapshot = BackupDir {
 302             group: item.backup,
 303             time: item.last_backup,
 304         };
 305         Ok(snapshot.to_string())
 306     };
 307
 308     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 309         let item = GroupListItem::deserialize(record)?;
 310         Ok(pbs_tools::format::render_backup_file_list(&item.files))
 311     };
 312
 313     let options = default_table_format_options()
 314         .sortby("backup-type", false)
 315         .sortby("backup-id", false)
 316         .column(
 317             ColumnConfig::new("backup-id")
 318                 .renderer(render_group_path)
 319                 .header("group"),
 320         )
 321         .column(
 322             ColumnConfig::new("last-backup")
 323                 .renderer(render_last_backup)
 324                 .header("last snapshot")
 325                 .right_align(false),
 326         )
 327         .column(ColumnConfig::new("backup-count"))
 328         .column(ColumnConfig::new("files").renderer(render_files));
 329
 330     let mut data: Value = result["data"].take();
 331
 332     let return_type = &pbs_api_types::ADMIN_DATASTORE_LIST_GROUPS_RETURN_TYPE;
 333
 334     format_and_print_result_full(&mut data, return_type, &output_format, &options);
 335
 336     Ok(Value::Null)
 337 }
 338
 339 fn merge_group_into(to: &mut serde_json::Map<String, Value>, group: BackupGroup) {
 340     match serde_json::to_value(group).unwrap() {
 341         Value::Object(group) => to.extend(group),
 342         _ => unreachable!(),
 343     }
 344 }
 345
 346 #[api(
 347    input: {
 348         properties: {
 349             repository: {
 350                 schema: REPO_URL_SCHEMA,
 351                 optional: true,
 352             },
 353             group: {
 354                 type: String,
 355                 description: "Backup group.",
 356             },
 357             "new-owner": {
 358                 type: Authid,
 359             },
 360         }
 361    }
 362 )]
 363 /// Change owner of a backup group
 364 async fn change_backup_owner(group: String, mut param: Value) -> Result<(), Error> {
 365     let repo = extract_repository_from_value(&param)?;
 366
 367     let client = connect(&repo)?;
 368
 369     param.as_object_mut().unwrap().remove("repository");
 370
 371     let group: BackupGroup = group.parse()?;
 372
 373     merge_group_into(param.as_object_mut().unwrap(), group);
 374
 375     let path = format!("api2/json/admin/datastore/{}/change-owner", repo.store());
 376     client.post(&path, Some(param)).await?;
 377
 378     record_repository(&repo);
 379
 380     Ok(())
 381 }
 382
 383 #[api(
 384    input: {
 385         properties: {
 386             repository: {
 387                 schema: REPO_URL_SCHEMA,
 388                 optional: true,
 389             },
 390         }
 391    }
 392 )]
 393 /// Try to login. If successful, store ticket.
 394 async fn api_login(param: Value) -> Result<Value, Error> {
 395     let repo = extract_repository_from_value(&param)?;
 396
 397     let client = connect(&repo)?;
 398     client.login().await?;
 399
 400     record_repository(&repo);
 401
 402     Ok(Value::Null)
 403 }
 404
 405 #[api(
 406    input: {
 407         properties: {
 408             repository: {
 409                 schema: REPO_URL_SCHEMA,
 410                 optional: true,
 411             },
 412         }
 413    }
 414 )]
 415 /// Logout (delete stored ticket).
 416 fn api_logout(param: Value) -> Result<Value, Error> {
 417     let repo = extract_repository_from_value(&param)?;
 418
 419     delete_ticket_info("proxmox-backup", repo.host(), repo.user())?;
 420
 421     Ok(Value::Null)
 422 }
 423
 424 #[api(
 425    input: {
 426         properties: {
 427             repository: {
 428                 schema: REPO_URL_SCHEMA,
 429                 optional: true,
 430             },
 431             "output-format": {
 432                 schema: OUTPUT_FORMAT,
 433                 optional: true,
 434             },
 435         }
 436    }
 437 )]
 438 /// Show client and optional server version
 439 async fn api_version(param: Value) -> Result<(), Error> {
 440     let output_format = get_output_format(&param);
 441
 442     let mut version_info = json!({
 443         "client": {
 444             "version": pbs_buildcfg::PROXMOX_PKG_VERSION,
 445             "release": pbs_buildcfg::PROXMOX_PKG_RELEASE,
 446             "repoid": pbs_buildcfg::PROXMOX_PKG_REPOID,
 447         }
 448     });
 449
 450     let repo = extract_repository_from_value(&param);
 451     if let Ok(repo) = repo {
 452         let client = connect(&repo)?;
 453
 454         match client.get("api2/json/version", None).await {
 455             Ok(mut result) => version_info["server"] = result["data"].take(),
 456             Err(e) => eprintln!("could not connect to server - {}", e),
 457         }
 458     }
 459     if output_format == "text" {
 460         println!(
 461             "client version: {}.{}",
 462             pbs_buildcfg::PROXMOX_PKG_VERSION,
 463             pbs_buildcfg::PROXMOX_PKG_RELEASE,
 464         );
 465         if let Some(server) = version_info["server"].as_object() {
 466             let server_version = server["version"].as_str().unwrap();
 467             let server_release = server["release"].as_str().unwrap();
 468             println!("server version: {}.{}", server_version, server_release);
 469         }
 470     } else {
 471         format_and_print_result(&version_info, &output_format);
 472     }
 473
 474     Ok(())
 475 }
 476
 477 #[api(
 478     input: {
 479         properties: {
 480             repository: {
 481                 schema: REPO_URL_SCHEMA,
 482                 optional: true,
 483             },
 484             "output-format": {
 485                 schema: OUTPUT_FORMAT,
 486                 optional: true,
 487             },
 488         },
 489     },
 490 )]
 491 /// Start garbage collection for a specific repository.
 492 async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 493     let repo = extract_repository_from_value(&param)?;
 494
 495     let output_format = get_output_format(&param);
 496
 497     let client = connect(&repo)?;
 498
 499     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 500
 501     let result = client.post(&path, None).await?;
 502
 503     record_repository(&repo);
 504
 505     view_task_result(&client, result, &output_format).await?;
 506
 507     Ok(Value::Null)
 508 }
 509
 510 struct CatalogUploadResult {
 511     catalog_writer: Arc<Mutex<CatalogWriter<TokioWriterAdapter<StdChannelWriter<Error>>>>>,
 512     result: tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>,
 513 }
 514
 515 fn spawn_catalog_upload(
 516     client: Arc<BackupWriter>,
 517     encrypt: bool,
 518 ) -> Result<CatalogUploadResult, Error> {
 519     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
 520     let catalog_stream = proxmox_async::blocking::StdChannelStream(catalog_rx);
 521     let catalog_chunk_size = 512 * 1024;
 522     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 523
 524     let catalog_writer = Arc::new(Mutex::new(CatalogWriter::new(TokioWriterAdapter::new(
 525         StdChannelWriter::new(catalog_tx),
 526     ))?));
 527
 528     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 529
 530     let upload_options = UploadOptions {
 531         encrypt,
 532         compress: true,
 533         ..UploadOptions::default()
 534     };
 535
 536     tokio::spawn(async move {
 537         let catalog_upload_result = client
 538             .upload_stream(CATALOG_NAME, catalog_chunk_stream, upload_options)
 539             .await;
 540
 541         if let Err(ref err) = catalog_upload_result {
 542             eprintln!("catalog upload error - {}", err);
 543             client.cancel();
 544         }
 545
 546         let _ = catalog_result_tx.send(catalog_upload_result);
 547     });
 548
 549     Ok(CatalogUploadResult {
 550         catalog_writer,
 551         result: catalog_result_rx,
 552     })
 553 }
 554
 555 #[api(
 556    input: {
 557        properties: {
 558            backupspec: {
 559                type: Array,
 560                description: "List of backup source specifications ([<label.ext>:<path>] ...)",
 561                items: {
 562                    schema: BACKUP_SOURCE_SCHEMA,
 563                }
 564            },
 565            repository: {
 566                schema: REPO_URL_SCHEMA,
 567                optional: true,
 568            },
 569            "include-dev": {
 570                description: "Include mountpoints with same st_dev number (see ``man fstat``) as specified files.",
 571                optional: true,
 572                items: {
 573                    type: String,
 574                    description: "Path to file.",
 575                }
 576            },
 577            "all-file-systems": {
 578                type: Boolean,
 579                description: "Include all mounted subdirectories.",
 580                optional: true,
 581                default: false,
 582            },
 583            keyfile: {
 584                schema: KEYFILE_SCHEMA,
 585                optional: true,
 586            },
 587            "keyfd": {
 588                schema: KEYFD_SCHEMA,
 589                optional: true,
 590            },
 591            "master-pubkey-file": {
 592                schema: MASTER_PUBKEY_FILE_SCHEMA,
 593                optional: true,
 594            },
 595            "master-pubkey-fd": {
 596                schema: MASTER_PUBKEY_FD_SCHEMA,
 597                optional: true,
 598            },
 599            "crypt-mode": {
 600                type: CryptMode,
 601                optional: true,
 602            },
 603            "skip-lost-and-found": {
 604                type: Boolean,
 605                description: "Skip lost+found directory.",
 606                optional: true,
 607                default: false,
 608            },
 609            "ns": {
 610                schema: BACKUP_NAMESPACE_SCHEMA,
 611                optional: true,
 612            },
 613            "backup-type": {
 614                schema: BACKUP_TYPE_SCHEMA,
 615                optional: true,
 616            },
 617            "backup-id": {
 618                schema: BACKUP_ID_SCHEMA,
 619                optional: true,
 620            },
 621            "backup-time": {
 622                schema: BACKUP_TIME_SCHEMA,
 623                optional: true,
 624            },
 625            "chunk-size": {
 626                schema: CHUNK_SIZE_SCHEMA,
 627                optional: true,
 628            },
 629            rate: {
 630                schema: TRAFFIC_CONTROL_RATE_SCHEMA,
 631                optional: true,
 632            },
 633            burst: {
 634                schema: TRAFFIC_CONTROL_BURST_SCHEMA,
 635                optional: true,
 636            },
 637            "exclude": {
 638                type: Array,
 639                description: "List of paths or patterns for matching files to exclude.",
 640                optional: true,
 641                items: {
 642                    type: String,
 643                    description: "Path or match pattern.",
 644                 }
 645            },
 646            "entries-max": {
 647                type: Integer,
 648                description: "Max number of entries to hold in memory.",
 649                optional: true,
 650                default: pbs_client::pxar::ENCODER_MAX_ENTRIES as isize,
 651            },
 652            "verbose": {
 653                type: Boolean,
 654                description: "Verbose output.",
 655                optional: true,
 656                default: false,
 657            },
 658            "dry-run": {
 659                type: Boolean,
 660                description: "Just show what backup would do, but do not upload anything.",
 661                optional: true,
 662                default: false,
 663            },
 664        }
 665    }
 666 )]
 667 /// Create (host) backup.
 668 async fn create_backup(
 669     param: Value,
 670     all_file_systems: bool,
 671     skip_lost_and_found: bool,
 672     dry_run: bool,
 673     verbose: bool,
 674     _info: &ApiMethod,
 675     _rpcenv: &mut dyn RpcEnvironment,
 676 ) -> Result<Value, Error> {
 677     let repo = extract_repository_from_value(&param)?;
 678
 679     let backupspec_list = json::required_array_param(&param, "backupspec")?;
 680
 681     let backup_time_opt = param["backup-time"].as_i64();
 682
 683     let chunk_size_opt = param["chunk-size"].as_u64().map(|v| (v * 1024) as usize);
 684
 685     if let Some(size) = chunk_size_opt {
 686         verify_chunk_size(size)?;
 687     }
 688
 689     let rate = match param["rate"].as_str() {
 690         Some(s) => Some(s.parse::<HumanByte>()?),
 691         None => None,
 692     };
 693     let burst = match param["burst"].as_str() {
 694         Some(s) => Some(s.parse::<HumanByte>()?),
 695         None => None,
 696     };
 697
 698     let rate_limit = RateLimitConfig::with_same_inout(rate, burst);
 699
 700     let crypto = crypto_parameters(&param)?;
 701
 702     let backup_id = param["backup-id"]
 703         .as_str()
 704         .unwrap_or(proxmox_sys::nodename());
 705
 706     let backup_ns = optional_ns_param(&param)?;
 707
 708     let backup_type: BackupType = param["backup-type"].as_str().unwrap_or("host").parse()?;
 709
 710     let include_dev = param["include-dev"].as_array();
 711
 712     let entries_max = param["entries-max"]
 713         .as_u64()
 714         .unwrap_or(pbs_client::pxar::ENCODER_MAX_ENTRIES as u64);
 715
 716     let empty = Vec::new();
 717     let exclude_args = param["exclude"].as_array().unwrap_or(&empty);
 718
 719     let mut pattern_list = Vec::with_capacity(exclude_args.len());
 720     for entry in exclude_args {
 721         let entry = entry
 722             .as_str()
 723             .ok_or_else(|| format_err!("Invalid pattern string slice"))?;
 724         pattern_list.push(
 725             MatchEntry::parse_pattern(entry, PatternFlag::PATH_NAME, MatchType::Exclude)
 726                 .map_err(|err| format_err!("invalid exclude pattern entry: {}", err))?,
 727         );
 728     }
 729
 730     let mut devices = if all_file_systems {
 731         None
 732     } else {
 733         Some(HashSet::new())
 734     };
 735
 736     if let Some(include_dev) = include_dev {
 737         if all_file_systems {
 738             bail!("option 'all-file-systems' conflicts with option 'include-dev'");
 739         }
 740
 741         let mut set = HashSet::new();
 742         for path in include_dev {
 743             let path = path.as_str().unwrap();
 744             let stat = nix::sys::stat::stat(path)
 745                 .map_err(|err| format_err!("fstat {:?} failed - {}", path, err))?;
 746             set.insert(stat.st_dev);
 747         }
 748         devices = Some(set);
 749     }
 750
 751     let mut upload_list = vec![];
 752     let mut target_set = HashSet::new();
 753
 754     for backupspec in backupspec_list {
 755         let spec = parse_backup_specification(backupspec.as_str().unwrap())?;
 756         let filename = &spec.config_string;
 757         let target = &spec.archive_name;
 758
 759         if target_set.contains(target) {
 760             bail!("got target twice: '{}'", target);
 761         }
 762         target_set.insert(target.to_string());
 763
 764         use std::os::unix::fs::FileTypeExt;
 765
 766         let metadata = std::fs::metadata(filename)
 767             .map_err(|err| format_err!("unable to access '{}' - {}", filename, err))?;
 768         let file_type = metadata.file_type();
 769
 770         match spec.spec_type {
 771             BackupSpecificationType::PXAR => {
 772                 if !file_type.is_dir() {
 773                     bail!("got unexpected file type (expected directory)");
 774                 }
 775                 upload_list.push((
 776                     BackupSpecificationType::PXAR,
 777                     filename.to_owned(),
 778                     format!("{}.didx", target),
 779                     0,
 780                 ));
 781             }
 782             BackupSpecificationType::IMAGE => {
 783                 if !(file_type.is_file() || file_type.is_block_device()) {
 784                     bail!("got unexpected file type (expected file or block device)");
 785                 }
 786
 787                 let size = image_size(&PathBuf::from(filename))?;
 788
 789                 if size == 0 {
 790                     bail!("got zero-sized file '{}'", filename);
 791                 }
 792
 793                 upload_list.push((
 794                     BackupSpecificationType::IMAGE,
 795                     filename.to_owned(),
 796                     format!("{}.fidx", target),
 797                     size,
 798                 ));
 799             }
 800             BackupSpecificationType::CONFIG => {
 801                 if !file_type.is_file() {
 802                     bail!("got unexpected file type (expected regular file)");
 803                 }
 804                 upload_list.push((
 805                     BackupSpecificationType::CONFIG,
 806                     filename.to_owned(),
 807                     format!("{}.blob", target),
 808                     metadata.len(),
 809                 ));
 810             }
 811             BackupSpecificationType::LOGFILE => {
 812                 if !file_type.is_file() {
 813                     bail!("got unexpected file type (expected regular file)");
 814                 }
 815                 upload_list.push((
 816                     BackupSpecificationType::LOGFILE,
 817                     filename.to_owned(),
 818                     format!("{}.blob", target),
 819                     metadata.len(),
 820                 ));
 821             }
 822         }
 823     }
 824
 825     let backup_time = backup_time_opt.unwrap_or_else(epoch_i64);
 826
 827     let client = connect_rate_limited(&repo, rate_limit)?;
 828     record_repository(&repo);
 829
 830     let snapshot = BackupDir::from((backup_type, backup_id.to_owned(), backup_time));
 831     if backup_ns.is_root() {
 832         println!("Starting backup: {snapshot}");
 833     } else {
 834         println!("Starting backup: [{backup_ns}]:{snapshot}");
 835     }
 836
 837     println!("Client name: {}", proxmox_sys::nodename());
 838
 839     let start_time = std::time::Instant::now();
 840
 841     println!(
 842         "Starting backup protocol: {}",
 843         strftime_local("%c", epoch_i64())?
 844     );
 845
 846     let (crypt_config, rsa_encrypted_key) = match crypto.enc_key {
 847         None => (None, None),
 848         Some(key_with_source) => {
 849             println!(
 850                 "{}",
 851                 format_key_source(&key_with_source.source, "encryption")
 852             );
 853
 854             let (key, created, fingerprint) =
 855                 decrypt_key(&key_with_source.key, &get_encryption_key_password)?;
 856             println!("Encryption key fingerprint: {}", fingerprint);
 857
 858             let crypt_config = CryptConfig::new(key)?;
 859
 860             match crypto.master_pubkey {
 861                 Some(pem_with_source) => {
 862                     println!("{}", format_key_source(&pem_with_source.source, "master"));
 863
 864                     let rsa = openssl::rsa::Rsa::public_key_from_pem(&pem_with_source.key)?;
 865
 866                     let mut key_config = KeyConfig::without_password(key)?;
 867                     key_config.created = created; // keep original value
 868
 869                     let enc_key = rsa_encrypt_key_config(rsa, &key_config)?;
 870
 871                     (Some(Arc::new(crypt_config)), Some(enc_key))
 872                 }
 873                 _ => (Some(Arc::new(crypt_config)), None),
 874             }
 875         }
 876     };
 877
 878     let client = BackupWriter::start(
 879         client,
 880         crypt_config.clone(),
 881         repo.store(),
 882         &backup_ns,
 883         &snapshot,
 884         verbose,
 885         false,
 886     )
 887     .await?;
 888
 889     let download_previous_manifest = match client.previous_backup_time().await {
 890         Ok(Some(backup_time)) => {
 891             println!(
 892                 "Downloading previous manifest ({})",
 893                 strftime_local("%c", backup_time)?
 894             );
 895             true
 896         }
 897         Ok(None) => {
 898             println!("No previous manifest available.");
 899             false
 900         }
 901         Err(_) => {
 902             // Fallback for outdated server, TODO remove/bubble up with 2.0
 903             true
 904         }
 905     };
 906
 907     let previous_manifest = if download_previous_manifest {
 908         match client.download_previous_manifest().await {
 909             Ok(previous_manifest) => {
 910                 match previous_manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref)) {
 911                     Ok(()) => Some(Arc::new(previous_manifest)),
 912                     Err(err) => {
 913                         println!("Couldn't re-use previous manifest - {}", err);
 914                         None
 915                     }
 916                 }
 917             }
 918             Err(err) => {
 919                 println!("Couldn't download previous manifest - {}", err);
 920                 None
 921             }
 922         }
 923     } else {
 924         None
 925     };
 926
 927     let mut manifest = BackupManifest::new(snapshot);
 928
 929     let mut catalog = None;
 930     let mut catalog_result_rx = None;
 931
 932     let log_file = |desc: &str, file: &str, target: &str| {
 933         let what = if dry_run { "Would upload" } else { "Upload" };
 934         println!("{} {} '{}' to '{}' as {}", what, desc, file, repo, target);
 935     };
 936
 937     for (backup_type, filename, target, size) in upload_list {
 938         match (backup_type, dry_run) {
 939             // dry-run
 940             (BackupSpecificationType::CONFIG, true) => log_file("config file", &filename, &target),
 941             (BackupSpecificationType::LOGFILE, true) => log_file("log file", &filename, &target),
 942             (BackupSpecificationType::PXAR, true) => log_file("directory", &filename, &target),
 943             (BackupSpecificationType::IMAGE, true) => log_file("image", &filename, &target),
 944             // no dry-run
 945             (BackupSpecificationType::CONFIG, false) => {
 946                 let upload_options = UploadOptions {
 947                     compress: true,
 948                     encrypt: crypto.mode == CryptMode::Encrypt,
 949                     ..UploadOptions::default()
 950                 };
 951
 952                 log_file("config file", &filename, &target);
 953                 let stats = client
 954                     .upload_blob_from_file(&filename, &target, upload_options)
 955                     .await?;
 956                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 957             }
 958             (BackupSpecificationType::LOGFILE, false) => {
 959                 // fixme: remove - not needed anymore ?
 960                 let upload_options = UploadOptions {
 961                     compress: true,
 962                     encrypt: crypto.mode == CryptMode::Encrypt,
 963                     ..UploadOptions::default()
 964                 };
 965
 966                 log_file("log file", &filename, &target);
 967                 let stats = client
 968                     .upload_blob_from_file(&filename, &target, upload_options)
 969                     .await?;
 970                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 971             }
 972             (BackupSpecificationType::PXAR, false) => {
 973                 // start catalog upload on first use
 974                 if catalog.is_none() {
 975                     let catalog_upload_res =
 976                         spawn_catalog_upload(client.clone(), crypto.mode == CryptMode::Encrypt)?;
 977                     catalog = Some(catalog_upload_res.catalog_writer);
 978                     catalog_result_rx = Some(catalog_upload_res.result);
 979                 }
 980                 let catalog = catalog.as_ref().unwrap();
 981
 982                 log_file("directory", &filename, &target);
 983                 catalog
 984                     .lock()
 985                     .unwrap()
 986                     .start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
 987
 988                 let pxar_options = pbs_client::pxar::PxarCreateOptions {
 989                     device_set: devices.clone(),
 990                     patterns: pattern_list.clone(),
 991                     entries_max: entries_max as usize,
 992                     skip_lost_and_found,
 993                     verbose,
 994                 };
 995
 996                 let upload_options = UploadOptions {
 997                     previous_manifest: previous_manifest.clone(),
 998                     compress: true,
 999                     encrypt: crypto.mode == CryptMode::Encrypt,
1000                     ..UploadOptions::default()
1001                 };
1002
1003                 let stats = backup_directory(
1004                     &client,
1005                     &filename,
1006                     &target,
1007                     chunk_size_opt,
1008                     catalog.clone(),
1009                     pxar_options,
1010                     upload_options,
1011                 )
1012                 .await?;
1013                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
1014                 catalog.lock().unwrap().end_directory()?;
1015             }
1016             (BackupSpecificationType::IMAGE, false) => {
1017                 log_file("image", &filename, &target);
1018
1019                 let upload_options = UploadOptions {
1020                     previous_manifest: previous_manifest.clone(),
1021                     fixed_size: Some(size),
1022                     compress: true,
1023                     encrypt: crypto.mode == CryptMode::Encrypt,
1024                 };
1025
1026                 let stats =
1027                     backup_image(&client, &filename, &target, chunk_size_opt, upload_options)
1028                         .await?;
1029                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
1030             }
1031         }
1032     }
1033
1034     if dry_run {
1035         println!("dry-run: no upload happend");
1036         return Ok(Value::Null);
1037     }
1038
1039     // finalize and upload catalog
1040     if let Some(catalog) = catalog {
1041         let mutex = Arc::try_unwrap(catalog)
1042             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
1043         let mut catalog = mutex.into_inner().unwrap();
1044
1045         catalog.finish()?;
1046
1047         drop(catalog); // close upload stream
1048
1049         if let Some(catalog_result_rx) = catalog_result_rx {
1050             let stats = catalog_result_rx.await??;
1051             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypto.mode)?;
1052         }
1053     }
1054
1055     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
1056         let target = ENCRYPTED_KEY_BLOB_NAME;
1057         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
1058         let options = UploadOptions {
1059             compress: false,
1060             encrypt: false,
1061             ..UploadOptions::default()
1062         };
1063         let stats = client
1064             .upload_blob_from_data(rsa_encrypted_key, target, options)
1065             .await?;
1066         manifest.add_file(target.to_string(), stats.size, stats.csum, crypto.mode)?;
1067     }
1068     // create manifest (index.json)
1069     // manifests are never encrypted, but include a signature
1070     let manifest = manifest
1071         .to_string(crypt_config.as_ref().map(Arc::as_ref))
1072         .map_err(|err| format_err!("unable to format manifest - {}", err))?;
1073
1074     if verbose {
1075         println!("Upload index.json to '{}'", repo)
1076     };
1077     let options = UploadOptions {
1078         compress: true,
1079         encrypt: false,
1080         ..UploadOptions::default()
1081     };
1082     client
1083         .upload_blob_from_data(manifest.into_bytes(), MANIFEST_BLOB_NAME, options)
1084         .await?;
1085
1086     client.finish().await?;
1087
1088     let end_time = std::time::Instant::now();
1089     let elapsed = end_time.duration_since(start_time);
1090     println!("Duration: {:.2}s", elapsed.as_secs_f64());
1091
1092     println!("End Time: {}", strftime_local("%c", epoch_i64())?);
1093
1094     Ok(Value::Null)
1095 }
1096
1097 async fn dump_image<W: Write>(
1098     client: Arc<BackupReader>,
1099     crypt_config: Option<Arc<CryptConfig>>,
1100     crypt_mode: CryptMode,
1101     index: FixedIndexReader,
1102     mut writer: W,
1103     verbose: bool,
1104 ) -> Result<(), Error> {
1105     let most_used = index.find_most_used_chunks(8);
1106
1107     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, crypt_mode, most_used);
1108
1109     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
1110     // and thus slows down reading. Instead, directly use RemoteChunkReader
1111     let mut per = 0;
1112     let mut bytes = 0;
1113     let start_time = std::time::Instant::now();
1114
1115     for pos in 0..index.index_count() {
1116         let digest = index.index_digest(pos).unwrap();
1117         let raw_data = chunk_reader.read_chunk(digest).await?;
1118         writer.write_all(&raw_data)?;
1119         bytes += raw_data.len();
1120         if verbose {
1121             let next_per = ((pos + 1) * 100) / index.index_count();
1122             if per != next_per {
1123                 eprintln!(
1124                     "progress {}% (read {} bytes, duration {} sec)",
1125                     next_per,
1126                     bytes,
1127                     start_time.elapsed().as_secs()
1128                 );
1129                 per = next_per;
1130             }
1131         }
1132     }
1133
1134     let end_time = std::time::Instant::now();
1135     let elapsed = end_time.duration_since(start_time);
1136     eprintln!(
1137         "restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1138         bytes,
1139         elapsed.as_secs_f64(),
1140         bytes as f64 / (1024.0 * 1024.0 * elapsed.as_secs_f64())
1141     );
1142
1143     Ok(())
1144 }
1145
1146 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1147     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1148         (name.into(), archive_type(name).unwrap())
1149     } else if name.ends_with(".pxar") {
1150         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1151     } else if name.ends_with(".img") {
1152         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1153     } else {
1154         (format!("{}.blob", name), ArchiveType::Blob)
1155     }
1156 }
1157
1158 #[api(
1159     input: {
1160         properties: {
1161             repository: {
1162                 schema: REPO_URL_SCHEMA,
1163                 optional: true,
1164             },
1165             ns: {
1166                 type: BackupNamespace,
1167                 optional: true,
1168             },
1169             snapshot: {
1170                 type: String,
1171                 description: "Group/Snapshot path.",
1172             },
1173             "archive-name": {
1174                 description: "Backup archive name.",
1175                 type: String,
1176             },
1177             target: {
1178                 type: String,
1179                 description: r###"Target directory path. Use '-' to write to standard output.
1180
1181 We do not extract '.pxar' archives when writing to standard output.
1182
1183 "###
1184             },
1185             rate: {
1186                 schema: TRAFFIC_CONTROL_RATE_SCHEMA,
1187                 optional: true,
1188             },
1189             burst: {
1190                 schema: TRAFFIC_CONTROL_BURST_SCHEMA,
1191                 optional: true,
1192             },
1193             "allow-existing-dirs": {
1194                 type: Boolean,
1195                 description: "Do not fail if directories already exists.",
1196                 optional: true,
1197             },
1198             keyfile: {
1199                 schema: KEYFILE_SCHEMA,
1200                 optional: true,
1201             },
1202             "keyfd": {
1203                 schema: KEYFD_SCHEMA,
1204                 optional: true,
1205             },
1206             "crypt-mode": {
1207                 type: CryptMode,
1208                 optional: true,
1209             },
1210         }
1211     }
1212 )]
1213 /// Restore backup repository.
1214 async fn restore(param: Value) -> Result<Value, Error> {
1215     let repo = extract_repository_from_value(&param)?;
1216
1217     let verbose = param["verbose"].as_bool().unwrap_or(false);
1218
1219     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1220
1221     let archive_name = json::required_string_param(&param, "archive-name")?;
1222
1223     let rate = match param["rate"].as_str() {
1224         Some(s) => Some(s.parse::<HumanByte>()?),
1225         None => None,
1226     };
1227     let burst = match param["burst"].as_str() {
1228         Some(s) => Some(s.parse::<HumanByte>()?),
1229         None => None,
1230     };
1231
1232     let rate_limit = RateLimitConfig::with_same_inout(rate, burst);
1233
1234     let client = connect_rate_limited(&repo, rate_limit)?;
1235     record_repository(&repo);
1236
1237     let ns = match param.get("ns") {
1238         Some(Value::String(ns)) => ns.parse()?,
1239         Some(_) => bail!("invalid namespace parameter"),
1240         None => BackupNamespace::root(),
1241     };
1242     let path = json::required_string_param(&param, "snapshot")?;
1243
1244     let backup_dir = dir_or_last_from_group(&client, &repo, &ns, &path).await?;
1245
1246     let target = json::required_string_param(&param, "target")?;
1247     let target = if target == "-" { None } else { Some(target) };
1248
1249     let crypto = crypto_parameters(&param)?;
1250
1251     let crypt_config = match crypto.enc_key {
1252         None => None,
1253         Some(ref key) => {
1254             let (key, _, _) =
1255                 decrypt_key(&key.key, &get_encryption_key_password).map_err(|err| {
1256                     eprintln!("{}", format_key_source(&key.source, "encryption"));
1257                     err
1258                 })?;
1259             Some(Arc::new(CryptConfig::new(key)?))
1260         }
1261     };
1262
1263     let client = BackupReader::start(
1264         client,
1265         crypt_config.clone(),
1266         repo.store(),
1267         &ns,
1268         &backup_dir,
1269         true,
1270     )
1271     .await?;
1272
1273     let (archive_name, archive_type) = parse_archive_type(archive_name);
1274
1275     let (manifest, backup_index_data) = client.download_manifest().await?;
1276
1277     if archive_name == ENCRYPTED_KEY_BLOB_NAME && crypt_config.is_none() {
1278         eprintln!("Restoring encrypted key blob without original key - skipping manifest fingerprint check!")
1279     } else {
1280         if manifest.signature.is_some() {
1281             if let Some(key) = &crypto.enc_key {
1282                 eprintln!("{}", format_key_source(&key.source, "encryption"));
1283             }
1284             if let Some(config) = &crypt_config {
1285                 eprintln!("Fingerprint: {}", Fingerprint::new(config.fingerprint()));
1286             }
1287         }
1288         manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref))?;
1289     }
1290
1291     if archive_name == MANIFEST_BLOB_NAME {
1292         if let Some(target) = target {
1293             replace_file(target, &backup_index_data, CreateOptions::new(), false)?;
1294         } else {
1295             let stdout = std::io::stdout();
1296             let mut writer = stdout.lock();
1297             writer
1298                 .write_all(&backup_index_data)
1299                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1300         }
1301
1302         return Ok(Value::Null);
1303     }
1304
1305     let file_info = manifest.lookup_file_info(&archive_name)?;
1306
1307     if archive_type == ArchiveType::Blob {
1308         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1309
1310         if let Some(target) = target {
1311             let mut writer = std::fs::OpenOptions::new()
1312                 .write(true)
1313                 .create(true)
1314                 .create_new(true)
1315                 .open(target)
1316                 .map_err(|err| {
1317                     format_err!("unable to create target file {:?} - {}", target, err)
1318                 })?;
1319             std::io::copy(&mut reader, &mut writer)?;
1320         } else {
1321             let stdout = std::io::stdout();
1322             let mut writer = stdout.lock();
1323             std::io::copy(&mut reader, &mut writer)
1324                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1325         }
1326     } else if archive_type == ArchiveType::DynamicIndex {
1327         let index = client
1328             .download_dynamic_index(&manifest, &archive_name)
1329             .await?;
1330
1331         let most_used = index.find_most_used_chunks(8);
1332
1333         let chunk_reader = RemoteChunkReader::new(
1334             client.clone(),
1335             crypt_config,
1336             file_info.chunk_crypt_mode(),
1337             most_used,
1338         );
1339
1340         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1341
1342         let options = pbs_client::pxar::PxarExtractOptions {
1343             match_list: &[],
1344             extract_match_default: true,
1345             allow_existing_dirs,
1346             on_error: None,
1347         };
1348
1349         if let Some(target) = target {
1350             pbs_client::pxar::extract_archive(
1351                 pxar::decoder::Decoder::from_std(reader)?,
1352                 Path::new(target),
1353                 pbs_client::pxar::Flags::DEFAULT,
1354                 |path| {
1355                     if verbose {
1356                         println!("{:?}", path);
1357                     }
1358                 },
1359                 options,
1360             )
1361             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1362         } else {
1363             let mut writer = std::fs::OpenOptions::new()
1364                 .write(true)
1365                 .open("/dev/stdout")
1366                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1367
1368             std::io::copy(&mut reader, &mut writer)
1369                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1370         }
1371     } else if archive_type == ArchiveType::FixedIndex {
1372         let index = client
1373             .download_fixed_index(&manifest, &archive_name)
1374             .await?;
1375
1376         let mut writer = if let Some(target) = target {
1377             std::fs::OpenOptions::new()
1378                 .write(true)
1379                 .create(true)
1380                 .create_new(true)
1381                 .open(target)
1382                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1383         } else {
1384             std::fs::OpenOptions::new()
1385                 .write(true)
1386                 .open("/dev/stdout")
1387                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1388         };
1389
1390         dump_image(
1391             client.clone(),
1392             crypt_config.clone(),
1393             file_info.chunk_crypt_mode(),
1394             index,
1395             &mut writer,
1396             verbose,
1397         )
1398         .await?;
1399     }
1400
1401     Ok(Value::Null)
1402 }
1403
1404 #[api(
1405     input: {
1406         properties: {
1407             "dry-run": {
1408                 type: bool,
1409                 optional: true,
1410                 description: "Just show what prune would do, but do not delete anything.",
1411             },
1412             group: {
1413                 type: String,
1414                 description: "Backup group",
1415             },
1416             "prune-options": {
1417                 type: PruneOptions,
1418                 flatten: true,
1419             },
1420             "output-format": {
1421                 schema: OUTPUT_FORMAT,
1422                 optional: true,
1423             },
1424             quiet: {
1425                 type: bool,
1426                 optional: true,
1427                 default: false,
1428                 description: "Minimal output - only show removals.",
1429             },
1430             repository: {
1431                 schema: REPO_URL_SCHEMA,
1432                 optional: true,
1433             },
1434         },
1435     },
1436 )]
1437 /// Prune a backup repository.
1438 async fn prune(
1439     dry_run: Option<bool>,
1440     group: String,
1441     prune_options: PruneOptions,
1442     quiet: bool,
1443     mut param: Value,
1444 ) -> Result<Value, Error> {
1445     let repo = extract_repository_from_value(&param)?;
1446
1447     let client = connect(&repo)?;
1448
1449     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1450
1451     let group: BackupGroup = group.parse()?;
1452
1453     let output_format = extract_output_format(&mut param);
1454
1455     let mut api_param = serde_json::to_value(prune_options)?;
1456     if let Some(dry_run) = dry_run {
1457         api_param["dry-run"] = dry_run.into();
1458     }
1459     merge_group_into(api_param.as_object_mut().unwrap(), group);
1460
1461     let mut result = client.post(&path, Some(api_param)).await?;
1462
1463     record_repository(&repo);
1464
1465     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1466         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1467         Ok(item.backup.to_string())
1468     };
1469
1470     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1471         Ok(match v.as_bool() {
1472             Some(true) => "keep",
1473             Some(false) => "remove",
1474             None => "unknown",
1475         }
1476         .to_string())
1477     };
1478
1479     let options = default_table_format_options()
1480         .sortby("backup-type", false)
1481         .sortby("backup-id", false)
1482         .sortby("backup-time", false)
1483         .column(
1484             ColumnConfig::new("backup-id")
1485                 .renderer(render_snapshot_path)
1486                 .header("snapshot"),
1487         )
1488         .column(
1489             ColumnConfig::new("backup-time")
1490                 .renderer(pbs_tools::format::render_epoch)
1491                 .header("date"),
1492         )
1493         .column(
1494             ColumnConfig::new("keep")
1495                 .renderer(render_prune_action)
1496                 .header("action"),
1497         );
1498
1499     let return_type = &pbs_api_types::ADMIN_DATASTORE_PRUNE_RETURN_TYPE;
1500
1501     let mut data = result["data"].take();
1502
1503     if quiet {
1504         let list: Vec<Value> = data
1505             .as_array()
1506             .unwrap()
1507             .iter()
1508             .filter(|item| item["keep"].as_bool() == Some(false))
1509             .cloned()
1510             .collect();
1511         data = list.into();
1512     }
1513
1514     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1515
1516     Ok(Value::Null)
1517 }
1518
1519 #[api(
1520    input: {
1521        properties: {
1522            repository: {
1523                schema: REPO_URL_SCHEMA,
1524                optional: true,
1525            },
1526            "output-format": {
1527                schema: OUTPUT_FORMAT,
1528                optional: true,
1529            },
1530        }
1531    },
1532     returns: {
1533         type: StorageStatus,
1534     },
1535 )]
1536 /// Get repository status.
1537 async fn status(param: Value) -> Result<Value, Error> {
1538     let repo = extract_repository_from_value(&param)?;
1539
1540     let output_format = get_output_format(&param);
1541
1542     let client = connect(&repo)?;
1543
1544     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1545
1546     let mut result = client.get(&path, None).await?;
1547     let mut data = result["data"].take();
1548
1549     record_repository(&repo);
1550
1551     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1552         let v = v.as_u64().unwrap();
1553         let total = record["total"].as_u64().unwrap();
1554         let roundup = total / 200;
1555         let per = ((v + roundup) * 100) / total;
1556         let info = format!(" ({} %)", per);
1557         Ok(format!("{} {:>8}", v, info))
1558     };
1559
1560     let options = default_table_format_options()
1561         .noheader(true)
1562         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1563         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1564         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1565
1566     let return_type = &API_METHOD_STATUS.returns;
1567
1568     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1569
1570     Ok(Value::Null)
1571 }
1572
1573 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
1574 /// async use!
1575 ///
1576 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
1577 /// so that we can properly access it from multiple threads simultaneously while not issuing
1578 /// duplicate simultaneous reads over http.
1579 pub struct BufferedDynamicReadAt {
1580     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
1581 }
1582
1583 impl BufferedDynamicReadAt {
1584     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
1585         Self {
1586             inner: Mutex::new(inner),
1587         }
1588     }
1589 }
1590
1591 impl ReadAt for BufferedDynamicReadAt {
1592     fn start_read_at<'a>(
1593         self: Pin<&'a Self>,
1594         _cx: &mut Context,
1595         buf: &'a mut [u8],
1596         offset: u64,
1597     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1598         MaybeReady::Ready(tokio::task::block_in_place(move || {
1599             let mut reader = self.inner.lock().unwrap();
1600             reader.seek(SeekFrom::Start(offset))?;
1601             reader.read(buf)
1602         }))
1603     }
1604
1605     fn poll_complete<'a>(
1606         self: Pin<&'a Self>,
1607         _op: ReadAtOperation<'a>,
1608     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1609         panic!("BufferedDynamicReadAt::start_read_at returned Pending");
1610     }
1611 }
1612
1613 fn main() {
1614     pbs_tools::setup_libc_malloc_opts();
1615
1616     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
1617         .arg_param(&["backupspec"])
1618         .completion_cb("repository", complete_repository)
1619         .completion_cb("backupspec", complete_backup_source)
1620         .completion_cb("keyfile", complete_file_name)
1621         .completion_cb("master-pubkey-file", complete_file_name)
1622         .completion_cb("chunk-size", complete_chunk_size);
1623
1624     let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
1625         .completion_cb("repository", complete_repository)
1626         .completion_cb("keyfile", complete_file_name);
1627
1628     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
1629         .completion_cb("repository", complete_repository);
1630
1631     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
1632         .completion_cb("repository", complete_repository);
1633
1634     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
1635         .arg_param(&["snapshot", "archive-name", "target"])
1636         .completion_cb("repository", complete_repository)
1637         .completion_cb("snapshot", complete_group_or_snapshot)
1638         .completion_cb("archive-name", complete_archive_name)
1639         .completion_cb("target", complete_file_name);
1640
1641     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
1642         .arg_param(&["group"])
1643         .completion_cb("group", complete_backup_group)
1644         .completion_cb("repository", complete_repository);
1645
1646     let status_cmd_def =
1647         CliCommand::new(&API_METHOD_STATUS).completion_cb("repository", complete_repository);
1648
1649     let login_cmd_def =
1650         CliCommand::new(&API_METHOD_API_LOGIN).completion_cb("repository", complete_repository);
1651
1652     let logout_cmd_def =
1653         CliCommand::new(&API_METHOD_API_LOGOUT).completion_cb("repository", complete_repository);
1654
1655     let version_cmd_def =
1656         CliCommand::new(&API_METHOD_API_VERSION).completion_cb("repository", complete_repository);
1657
1658     let change_owner_cmd_def = CliCommand::new(&API_METHOD_CHANGE_BACKUP_OWNER)
1659         .arg_param(&["group", "new-owner"])
1660         .completion_cb("group", complete_backup_group)
1661         .completion_cb("new-owner", complete_auth_id)
1662         .completion_cb("repository", complete_repository);
1663
1664     let cmd_def = CliCommandMap::new()
1665         .insert("backup", backup_cmd_def)
1666         .insert("garbage-collect", garbage_collect_cmd_def)
1667         .insert("list", list_cmd_def)
1668         .insert("login", login_cmd_def)
1669         .insert("logout", logout_cmd_def)
1670         .insert("prune", prune_cmd_def)
1671         .insert("restore", restore_cmd_def)
1672         .insert("snapshot", snapshot_mgtm_cli())
1673         .insert("status", status_cmd_def)
1674         .insert("key", key::cli())
1675         .insert("mount", mount_cmd_def())
1676         .insert("map", map_cmd_def())
1677         .insert("unmap", unmap_cmd_def())
1678         .insert("catalog", catalog_mgmt_cli())
1679         .insert("task", task_mgmt_cli())
1680         .insert("version", version_cmd_def)
1681         .insert("benchmark", benchmark_cmd_def)
1682         .insert("change-owner", change_owner_cmd_def)
1683         .insert("namespace", namespace::cli_map())
1684         .alias(&["files"], &["snapshot", "files"])
1685         .alias(&["forget"], &["snapshot", "forget"])
1686         .alias(&["upload-log"], &["snapshot", "upload-log"])
1687         .alias(&["snapshots"], &["snapshot", "list"]);
1688
1689     let rpcenv = CliEnvironment::new();
1690     run_cli_command(
1691         cmd_def,
1692         rpcenv,
1693         Some(|future| proxmox_async::runtime::main(future)),
1694     );
1695 }