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