src/bin/proxmox-backup-client.rs

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