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