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