src/bin/proxmox-backup-client.rs

   1 use std::collections::{HashSet, HashMap};
   2 use std::ffi::OsStr;
   3 use std::io::{self, Write, Seek, SeekFrom};
   4 use std::os::unix::fs::OpenOptionsExt;
   5 use std::os::unix::io::RawFd;
   6 use std::path::{Path, PathBuf};
   7 use std::pin::Pin;
   8 use std::sync::{Arc, Mutex};
   9 use std::task::{Context, Poll};
  10
  11 use anyhow::{bail, format_err, Error};
  12 use chrono::{Local, DateTime, Utc, TimeZone};
  13 use futures::future::FutureExt;
  14 use futures::select;
  15 use futures::stream::{StreamExt, TryStreamExt};
  16 use nix::unistd::{fork, ForkResult, pipe};
  17 use serde_json::{json, Value};
  18 use tokio::signal::unix::{signal, SignalKind};
  19 use tokio::sync::mpsc;
  20 use xdg::BaseDirectories;
  21
  22 use pathpatterns::{MatchEntry, MatchType, PatternFlag};
  23 use proxmox::{sortable, identity};
  24 use proxmox::tools::fs::{file_get_contents, file_get_json, replace_file, CreateOptions, image_size};
  25 use proxmox::sys::linux::tty;
  26 use proxmox::api::{ApiHandler, ApiMethod, RpcEnvironment};
  27 use proxmox::api::schema::*;
  28 use proxmox::api::cli::*;
  29 use proxmox::api::api;
  30
  31 use proxmox_backup::tools;
  32 use proxmox_backup::api2::types::*;
  33 use proxmox_backup::client::*;
  34 use proxmox_backup::pxar::catalog::*;
  35 use proxmox_backup::backup::{
  36     archive_type,
  37     encrypt_key_with_passphrase,
  38     load_and_decrypt_key,
  39     store_key_config,
  40     verify_chunk_size,
  41     ArchiveType,
  42     AsyncReadChunk,
  43     BackupDir,
  44     BackupGroup,
  45     BackupManifest,
  46     BufferedDynamicReader,
  47     CatalogReader,
  48     CatalogWriter,
  49     CATALOG_NAME,
  50     ChunkStream,
  51     CryptConfig,
  52     DataBlob,
  53     DynamicIndexReader,
  54     FixedChunkStream,
  55     FixedIndexReader,
  56     IndexFile,
  57     KeyConfig,
  58     MANIFEST_BLOB_NAME,
  59     Shell,
  60 };
  61
  62 const ENV_VAR_PBS_FINGERPRINT: &str = "PBS_FINGERPRINT";
  63 const ENV_VAR_PBS_PASSWORD: &str = "PBS_PASSWORD";
  64
  65
  66 const REPO_URL_SCHEMA: Schema = StringSchema::new("Repository URL.")
  67     .format(&BACKUP_REPO_URL)
  68     .max_length(256)
  69     .schema();
  70
  71 const KEYFILE_SCHEMA: Schema = StringSchema::new(
  72     "Path to encryption key. All data will be encrypted using this key.")
  73     .schema();
  74
  75 const CHUNK_SIZE_SCHEMA: Schema = IntegerSchema::new(
  76     "Chunk size in KB. Must be a power of 2.")
  77     .minimum(64)
  78     .maximum(4096)
  79     .default(4096)
  80     .schema();
  81
  82 fn get_default_repository() -> Option<String> {
  83     std::env::var("PBS_REPOSITORY").ok()
  84 }
  85
  86 fn extract_repository_from_value(
  87     param: &Value,
  88 ) -> Result<BackupRepository, Error> {
  89
  90     let repo_url = param["repository"]
  91         .as_str()
  92         .map(String::from)
  93         .or_else(get_default_repository)
  94         .ok_or_else(|| format_err!("unable to get (default) repository"))?;
  95
  96     let repo: BackupRepository = repo_url.parse()?;
  97
  98     Ok(repo)
  99 }
 100
 101 fn extract_repository_from_map(
 102     param: &HashMap<String, String>,
 103 ) -> Option<BackupRepository> {
 104
 105     param.get("repository")
 106         .map(String::from)
 107         .or_else(get_default_repository)
 108         .and_then(|repo_url| repo_url.parse::<BackupRepository>().ok())
 109 }
 110
 111 fn record_repository(repo: &BackupRepository) {
 112
 113     let base = match BaseDirectories::with_prefix("proxmox-backup") {
 114         Ok(v) => v,
 115         _ => return,
 116     };
 117
 118     // usually $HOME/.cache/proxmox-backup/repo-list
 119     let path = match base.place_cache_file("repo-list") {
 120         Ok(v) => v,
 121         _ => return,
 122     };
 123
 124     let mut data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
 125
 126     let repo = repo.to_string();
 127
 128     data[&repo] = json!{ data[&repo].as_i64().unwrap_or(0) + 1 };
 129
 130     let mut map = serde_json::map::Map::new();
 131
 132     loop {
 133         let mut max_used = 0;
 134         let mut max_repo = None;
 135         for (repo, count) in data.as_object().unwrap() {
 136             if map.contains_key(repo) { continue; }
 137             if let Some(count) = count.as_i64() {
 138                 if count > max_used {
 139                     max_used = count;
 140                     max_repo = Some(repo);
 141                 }
 142             }
 143         }
 144         if let Some(repo) = max_repo {
 145             map.insert(repo.to_owned(), json!(max_used));
 146         } else {
 147             break;
 148         }
 149         if map.len() > 10 { // store max. 10 repos
 150             break;
 151         }
 152     }
 153
 154     let new_data = json!(map);
 155
 156     let _ = replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new());
 157 }
 158
 159 fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
 160
 161     let mut result = vec![];
 162
 163     let base = match BaseDirectories::with_prefix("proxmox-backup") {
 164         Ok(v) => v,
 165         _ => return result,
 166     };
 167
 168     // usually $HOME/.cache/proxmox-backup/repo-list
 169     let path = match base.place_cache_file("repo-list") {
 170         Ok(v) => v,
 171         _ => return result,
 172     };
 173
 174     let data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
 175
 176     if let Some(map) = data.as_object() {
 177         for (repo, _count) in map {
 178             result.push(repo.to_owned());
 179         }
 180     }
 181
 182     result
 183 }
 184
 185 fn connect(server: &str, userid: &str) -> Result<HttpClient, Error> {
 186
 187     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
 188
 189     use std::env::VarError::*;
 190     let password = match std::env::var(ENV_VAR_PBS_PASSWORD) {
 191         Ok(p) => Some(p),
 192         Err(NotUnicode(_)) => bail!(format!("{} contains bad characters", ENV_VAR_PBS_PASSWORD)),
 193         Err(NotPresent) => None,
 194     };
 195
 196     let options = HttpClientOptions::new()
 197         .prefix(Some("proxmox-backup".to_string()))
 198         .password(password)
 199         .interactive(true)
 200         .fingerprint(fingerprint)
 201         .fingerprint_cache(true)
 202         .ticket_cache(true);
 203
 204     HttpClient::new(server, userid, options)
 205 }
 206
 207 async fn view_task_result(
 208     client: HttpClient,
 209     result: Value,
 210     output_format: &str,
 211 ) -> Result<(), Error> {
 212     let data = &result["data"];
 213     if output_format == "text" {
 214         if let Some(upid) = data.as_str() {
 215             display_task_log(client, upid, true).await?;
 216         }
 217     } else {
 218         format_and_print_result(&data, &output_format);
 219     }
 220
 221     Ok(())
 222 }
 223
 224 async fn api_datastore_list_snapshots(
 225     client: &HttpClient,
 226     store: &str,
 227     group: Option<BackupGroup>,
 228 ) -> Result<Value, Error> {
 229
 230     let path = format!("api2/json/admin/datastore/{}/snapshots", store);
 231
 232     let mut args = json!({});
 233     if let Some(group) = group {
 234         args["backup-type"] = group.backup_type().into();
 235         args["backup-id"] = group.backup_id().into();
 236     }
 237
 238     let mut result = client.get(&path, Some(args)).await?;
 239
 240     Ok(result["data"].take())
 241 }
 242
 243 async fn api_datastore_latest_snapshot(
 244     client: &HttpClient,
 245     store: &str,
 246     group: BackupGroup,
 247 ) -> Result<(String, String, DateTime<Utc>), Error> {
 248
 249     let list = api_datastore_list_snapshots(client, store, Some(group.clone())).await?;
 250     let mut list: Vec<SnapshotListItem> = serde_json::from_value(list)?;
 251
 252     if list.is_empty() {
 253         bail!("backup group {:?} does not contain any snapshots.", group.group_path());
 254     }
 255
 256     list.sort_unstable_by(|a, b| b.backup_time.cmp(&a.backup_time));
 257
 258     let backup_time = Utc.timestamp(list[0].backup_time, 0);
 259
 260     Ok((group.backup_type().to_owned(), group.backup_id().to_owned(), backup_time))
 261 }
 262
 263
 264 async fn backup_directory<P: AsRef<Path>>(
 265     client: &BackupWriter,
 266     dir_path: P,
 267     archive_name: &str,
 268     chunk_size: Option<usize>,
 269     device_set: Option<HashSet<u64>>,
 270     verbose: bool,
 271     skip_lost_and_found: bool,
 272     crypt_config: Option<Arc<CryptConfig>>,
 273     catalog: Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 274     exclude_pattern: Vec<MatchEntry>,
 275     entries_max: usize,
 276 ) -> Result<BackupStats, Error> {
 277
 278     let pxar_stream = PxarBackupStream::open(
 279         dir_path.as_ref(),
 280         device_set,
 281         verbose,
 282         skip_lost_and_found,
 283         catalog,
 284         exclude_pattern,
 285         entries_max,
 286     )?;
 287     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 288
 289     let (mut tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 290
 291     let stream = rx
 292         .map_err(Error::from);
 293
 294     // spawn chunker inside a separate task so that it can run parallel
 295     tokio::spawn(async move {
 296         while let Some(v) = chunk_stream.next().await {
 297             let _ = tx.send(v).await;
 298         }
 299     });
 300
 301     let stats = client
 302         .upload_stream(archive_name, stream, "dynamic", None, crypt_config)
 303         .await?;
 304
 305     Ok(stats)
 306 }
 307
 308 async fn backup_image<P: AsRef<Path>>(
 309     client: &BackupWriter,
 310     image_path: P,
 311     archive_name: &str,
 312     image_size: u64,
 313     chunk_size: Option<usize>,
 314     _verbose: bool,
 315     crypt_config: Option<Arc<CryptConfig>>,
 316 ) -> Result<BackupStats, Error> {
 317
 318     let path = image_path.as_ref().to_owned();
 319
 320     let file = tokio::fs::File::open(path).await?;
 321
 322     let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 323         .map_err(Error::from);
 324
 325     let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
 326
 327     let stats = client
 328         .upload_stream(archive_name, stream, "fixed", Some(image_size), crypt_config)
 329         .await?;
 330
 331     Ok(stats)
 332 }
 333
 334 #[api(
 335    input: {
 336         properties: {
 337             repository: {
 338                 schema: REPO_URL_SCHEMA,
 339                 optional: true,
 340             },
 341             "output-format": {
 342                 schema: OUTPUT_FORMAT,
 343                 optional: true,
 344             },
 345         }
 346    }
 347 )]
 348 /// List backup groups.
 349 async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 350
 351     let output_format = get_output_format(&param);
 352
 353     let repo = extract_repository_from_value(&param)?;
 354
 355     let client = connect(repo.host(), repo.user())?;
 356
 357     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 358
 359     let mut result = client.get(&path, None).await?;
 360
 361     record_repository(&repo);
 362
 363     let render_group_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 364         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 365         let group = BackupGroup::new(item.backup_type, item.backup_id);
 366         Ok(group.group_path().to_str().unwrap().to_owned())
 367     };
 368
 369     let render_last_backup = |_v: &Value, record: &Value| -> Result<String, Error> {
 370         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 371         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.last_backup);
 372         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 373     };
 374
 375     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 376         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 377         Ok(tools::format::render_backup_file_list(&item.files))
 378     };
 379
 380     let options = default_table_format_options()
 381         .sortby("backup-type", false)
 382         .sortby("backup-id", false)
 383         .column(ColumnConfig::new("backup-id").renderer(render_group_path).header("group"))
 384         .column(
 385             ColumnConfig::new("last-backup")
 386                 .renderer(render_last_backup)
 387                 .header("last snapshot")
 388                 .right_align(false)
 389         )
 390         .column(ColumnConfig::new("backup-count"))
 391         .column(ColumnConfig::new("files").renderer(render_files));
 392
 393     let mut data: Value = result["data"].take();
 394
 395     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_GROUPS;
 396
 397     format_and_print_result_full(&mut data, info, &output_format, &options);
 398
 399     Ok(Value::Null)
 400 }
 401
 402 #[api(
 403    input: {
 404         properties: {
 405             repository: {
 406                 schema: REPO_URL_SCHEMA,
 407                 optional: true,
 408             },
 409             group: {
 410                 type: String,
 411                 description: "Backup group.",
 412                 optional: true,
 413             },
 414             "output-format": {
 415                 schema: OUTPUT_FORMAT,
 416                 optional: true,
 417             },
 418         }
 419    }
 420 )]
 421 /// List backup snapshots.
 422 async fn list_snapshots(param: Value) -> Result<Value, Error> {
 423
 424     let repo = extract_repository_from_value(&param)?;
 425
 426     let output_format = get_output_format(&param);
 427
 428     let client = connect(repo.host(), repo.user())?;
 429
 430     let group = if let Some(path) = param["group"].as_str() {
 431         Some(BackupGroup::parse(path)?)
 432     } else {
 433         None
 434     };
 435
 436     let mut data = api_datastore_list_snapshots(&client, repo.store(), group).await?;
 437
 438     record_repository(&repo);
 439
 440     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 441         let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
 442         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time);
 443         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 444     };
 445
 446     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 447         let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
 448         Ok(tools::format::render_backup_file_list(&item.files))
 449     };
 450
 451     let options = default_table_format_options()
 452         .sortby("backup-type", false)
 453         .sortby("backup-id", false)
 454         .sortby("backup-time", false)
 455         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
 456         .column(ColumnConfig::new("size"))
 457         .column(ColumnConfig::new("files").renderer(render_files))
 458         ;
 459
 460     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOTS;
 461
 462     format_and_print_result_full(&mut data, info, &output_format, &options);
 463
 464     Ok(Value::Null)
 465 }
 466
 467 #[api(
 468    input: {
 469         properties: {
 470             repository: {
 471                 schema: REPO_URL_SCHEMA,
 472                 optional: true,
 473             },
 474             snapshot: {
 475                 type: String,
 476                 description: "Snapshot path.",
 477              },
 478         }
 479    }
 480 )]
 481 /// Forget (remove) backup snapshots.
 482 async fn forget_snapshots(param: Value) -> Result<Value, Error> {
 483
 484     let repo = extract_repository_from_value(&param)?;
 485
 486     let path = tools::required_string_param(&param, "snapshot")?;
 487     let snapshot = BackupDir::parse(path)?;
 488
 489     let mut client = connect(repo.host(), repo.user())?;
 490
 491     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
 492
 493     let result = client.delete(&path, Some(json!({
 494         "backup-type": snapshot.group().backup_type(),
 495         "backup-id": snapshot.group().backup_id(),
 496         "backup-time": snapshot.backup_time().timestamp(),
 497     }))).await?;
 498
 499     record_repository(&repo);
 500
 501     Ok(result)
 502 }
 503
 504 #[api(
 505    input: {
 506         properties: {
 507             repository: {
 508                 schema: REPO_URL_SCHEMA,
 509                 optional: true,
 510             },
 511         }
 512    }
 513 )]
 514 /// Try to login. If successful, store ticket.
 515 async fn api_login(param: Value) -> Result<Value, Error> {
 516
 517     let repo = extract_repository_from_value(&param)?;
 518
 519     let client = connect(repo.host(), repo.user())?;
 520     client.login().await?;
 521
 522     record_repository(&repo);
 523
 524     Ok(Value::Null)
 525 }
 526
 527 #[api(
 528    input: {
 529         properties: {
 530             repository: {
 531                 schema: REPO_URL_SCHEMA,
 532                 optional: true,
 533             },
 534         }
 535    }
 536 )]
 537 /// Logout (delete stored ticket).
 538 fn api_logout(param: Value) -> Result<Value, Error> {
 539
 540     let repo = extract_repository_from_value(&param)?;
 541
 542     delete_ticket_info("proxmox-backup", repo.host(), repo.user())?;
 543
 544     Ok(Value::Null)
 545 }
 546
 547 #[api(
 548    input: {
 549         properties: {
 550             repository: {
 551                 schema: REPO_URL_SCHEMA,
 552                 optional: true,
 553             },
 554             snapshot: {
 555                 type: String,
 556                 description: "Snapshot path.",
 557              },
 558         }
 559    }
 560 )]
 561 /// Dump catalog.
 562 async fn dump_catalog(param: Value) -> Result<Value, Error> {
 563
 564     let repo = extract_repository_from_value(&param)?;
 565
 566     let path = tools::required_string_param(&param, "snapshot")?;
 567     let snapshot = BackupDir::parse(path)?;
 568
 569     let keyfile = param["keyfile"].as_str().map(PathBuf::from);
 570
 571     let crypt_config = match keyfile {
 572         None => None,
 573         Some(path) => {
 574             let (key, _) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
 575             Some(Arc::new(CryptConfig::new(key)?))
 576         }
 577     };
 578
 579     let client = connect(repo.host(), repo.user())?;
 580
 581     let client = BackupReader::start(
 582         client,
 583         crypt_config.clone(),
 584         repo.store(),
 585         &snapshot.group().backup_type(),
 586         &snapshot.group().backup_id(),
 587         snapshot.backup_time(),
 588         true,
 589     ).await?;
 590
 591     let manifest = client.download_manifest().await?;
 592
 593     let index = client.download_dynamic_index(&manifest, CATALOG_NAME).await?;
 594
 595     let most_used = index.find_most_used_chunks(8);
 596
 597     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
 598
 599     let mut reader = BufferedDynamicReader::new(index, chunk_reader);
 600
 601     let mut catalogfile = std::fs::OpenOptions::new()
 602         .write(true)
 603         .read(true)
 604         .custom_flags(libc::O_TMPFILE)
 605         .open("/tmp")?;
 606
 607     std::io::copy(&mut reader, &mut catalogfile)
 608         .map_err(|err| format_err!("unable to download catalog - {}", err))?;
 609
 610     catalogfile.seek(SeekFrom::Start(0))?;
 611
 612     let mut catalog_reader = CatalogReader::new(catalogfile);
 613
 614     catalog_reader.dump()?;
 615
 616     record_repository(&repo);
 617
 618     Ok(Value::Null)
 619 }
 620
 621 #[api(
 622    input: {
 623         properties: {
 624             repository: {
 625                 schema: REPO_URL_SCHEMA,
 626                 optional: true,
 627             },
 628             snapshot: {
 629                 type: String,
 630                 description: "Snapshot path.",
 631              },
 632             "output-format": {
 633                 schema: OUTPUT_FORMAT,
 634                 optional: true,
 635             },
 636         }
 637    }
 638 )]
 639 /// List snapshot files.
 640 async fn list_snapshot_files(param: Value) -> Result<Value, Error> {
 641
 642     let repo = extract_repository_from_value(&param)?;
 643
 644     let path = tools::required_string_param(&param, "snapshot")?;
 645     let snapshot = BackupDir::parse(path)?;
 646
 647     let output_format = get_output_format(&param);
 648
 649     let client = connect(repo.host(), repo.user())?;
 650
 651     let path = format!("api2/json/admin/datastore/{}/files", repo.store());
 652
 653     let mut result = client.get(&path, Some(json!({
 654         "backup-type": snapshot.group().backup_type(),
 655         "backup-id": snapshot.group().backup_id(),
 656         "backup-time": snapshot.backup_time().timestamp(),
 657     }))).await?;
 658
 659     record_repository(&repo);
 660
 661     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOT_FILES;
 662
 663     let mut data: Value = result["data"].take();
 664
 665     let options = default_table_format_options();
 666
 667     format_and_print_result_full(&mut data, info, &output_format, &options);
 668
 669     Ok(Value::Null)
 670 }
 671
 672 #[api(
 673     input: {
 674         properties: {
 675             repository: {
 676                 schema: REPO_URL_SCHEMA,
 677                 optional: true,
 678             },
 679             "output-format": {
 680                 schema: OUTPUT_FORMAT,
 681                 optional: true,
 682             },
 683         },
 684     },
 685 )]
 686 /// Start garbage collection for a specific repository.
 687 async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 688
 689     let repo = extract_repository_from_value(&param)?;
 690
 691     let output_format = get_output_format(&param);
 692
 693     let mut client = connect(repo.host(), repo.user())?;
 694
 695     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 696
 697     let result = client.post(&path, None).await?;
 698
 699     record_repository(&repo);
 700
 701     view_task_result(client, result, &output_format).await?;
 702
 703     Ok(Value::Null)
 704 }
 705
 706 fn spawn_catalog_upload(
 707     client: Arc<BackupWriter>,
 708     crypt_config: Option<Arc<CryptConfig>>,
 709 ) -> Result<
 710         (
 711             Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 712             tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>
 713         ), Error>
 714 {
 715     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
 716     let catalog_stream = crate::tools::StdChannelStream(catalog_rx);
 717     let catalog_chunk_size = 512*1024;
 718     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 719
 720     let catalog = Arc::new(Mutex::new(CatalogWriter::new(crate::tools::StdChannelWriter::new(catalog_tx))?));
 721
 722     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 723
 724     tokio::spawn(async move {
 725         let catalog_upload_result = client
 726             .upload_stream(CATALOG_NAME, catalog_chunk_stream, "dynamic", None, crypt_config)
 727             .await;
 728
 729         if let Err(ref err) = catalog_upload_result {
 730             eprintln!("catalog upload error - {}", err);
 731             client.cancel();
 732         }
 733
 734         let _ = catalog_result_tx.send(catalog_upload_result);
 735     });
 736
 737     Ok((catalog, catalog_result_rx))
 738 }
 739
 740 #[api(
 741    input: {
 742        properties: {
 743            backupspec: {
 744                type: Array,
 745                description: "List of backup source specifications ([<label.ext>:<path>] ...)",
 746                items: {
 747                    schema: BACKUP_SOURCE_SCHEMA,
 748                }
 749            },
 750            repository: {
 751                schema: REPO_URL_SCHEMA,
 752                optional: true,
 753            },
 754            "include-dev": {
 755                description: "Include mountpoints with same st_dev number (see ``man fstat``) as specified files.",
 756                optional: true,
 757                items: {
 758                    type: String,
 759                    description: "Path to file.",
 760                }
 761            },
 762            keyfile: {
 763                schema: KEYFILE_SCHEMA,
 764                optional: true,
 765            },
 766            "skip-lost-and-found": {
 767                type: Boolean,
 768                description: "Skip lost+found directory.",
 769                optional: true,
 770            },
 771            "backup-type": {
 772                schema: BACKUP_TYPE_SCHEMA,
 773                optional: true,
 774            },
 775            "backup-id": {
 776                schema: BACKUP_ID_SCHEMA,
 777                optional: true,
 778            },
 779            "backup-time": {
 780                schema: BACKUP_TIME_SCHEMA,
 781                optional: true,
 782            },
 783            "chunk-size": {
 784                schema: CHUNK_SIZE_SCHEMA,
 785                optional: true,
 786            },
 787            "exclude": {
 788                type: Array,
 789                description: "List of paths or patterns for matching files to exclude.",
 790                optional: true,
 791                items: {
 792                    type: String,
 793                    description: "Path or match pattern.",
 794                 }
 795            },
 796            "entries-max": {
 797                type: Integer,
 798                description: "Max number of entries to hold in memory.",
 799                optional: true,
 800                default: proxmox_backup::pxar::ENCODER_MAX_ENTRIES as isize,
 801            },
 802            "verbose": {
 803                type: Boolean,
 804                description: "Verbose output.",
 805                optional: true,
 806            },
 807        }
 808    }
 809 )]
 810 /// Create (host) backup.
 811 async fn create_backup(
 812     param: Value,
 813     _info: &ApiMethod,
 814     _rpcenv: &mut dyn RpcEnvironment,
 815 ) -> Result<Value, Error> {
 816
 817     let repo = extract_repository_from_value(&param)?;
 818
 819     let backupspec_list = tools::required_array_param(&param, "backupspec")?;
 820
 821     let all_file_systems = param["all-file-systems"].as_bool().unwrap_or(false);
 822
 823     let skip_lost_and_found = param["skip-lost-and-found"].as_bool().unwrap_or(false);
 824
 825     let verbose = param["verbose"].as_bool().unwrap_or(false);
 826
 827     let backup_time_opt = param["backup-time"].as_i64();
 828
 829     let chunk_size_opt = param["chunk-size"].as_u64().map(|v| (v*1024) as usize);
 830
 831     if let Some(size) = chunk_size_opt {
 832         verify_chunk_size(size)?;
 833     }
 834
 835     let keyfile = param["keyfile"].as_str().map(PathBuf::from);
 836
 837     let backup_id = param["backup-id"].as_str().unwrap_or(&proxmox::tools::nodename());
 838
 839     let backup_type = param["backup-type"].as_str().unwrap_or("host");
 840
 841     let include_dev = param["include-dev"].as_array();
 842
 843     let entries_max = param["entries-max"].as_u64()
 844         .unwrap_or(proxmox_backup::pxar::ENCODER_MAX_ENTRIES as u64);
 845
 846     let empty = Vec::new();
 847     let exclude_args = param["exclude"].as_array().unwrap_or(&empty);
 848
 849     let mut pattern_list = Vec::with_capacity(exclude_args.len());
 850     for entry in exclude_args {
 851         let entry = entry.as_str().ok_or_else(|| format_err!("Invalid pattern string slice"))?;
 852         pattern_list.push(
 853             MatchEntry::parse_pattern(entry, PatternFlag::PATH_NAME, MatchType::Exclude)
 854                 .map_err(|err| format_err!("invalid exclude pattern entry: {}", err))?
 855         );
 856     }
 857
 858     let mut devices = if all_file_systems { None } else { Some(HashSet::new()) };
 859
 860     if let Some(include_dev) = include_dev {
 861         if all_file_systems {
 862             bail!("option 'all-file-systems' conflicts with option 'include-dev'");
 863         }
 864
 865         let mut set = HashSet::new();
 866         for path in include_dev {
 867             let path = path.as_str().unwrap();
 868             let stat = nix::sys::stat::stat(path)
 869                 .map_err(|err| format_err!("fstat {:?} failed - {}", path, err))?;
 870             set.insert(stat.st_dev);
 871         }
 872         devices = Some(set);
 873     }
 874
 875     let mut upload_list = vec![];
 876
 877     for backupspec in backupspec_list {
 878         let spec = parse_backup_specification(backupspec.as_str().unwrap())?;
 879         let filename = &spec.config_string;
 880         let target = &spec.archive_name;
 881
 882         use std::os::unix::fs::FileTypeExt;
 883
 884         let metadata = std::fs::metadata(filename)
 885             .map_err(|err| format_err!("unable to access '{}' - {}", filename, err))?;
 886         let file_type = metadata.file_type();
 887
 888         match spec.spec_type {
 889             BackupSpecificationType::PXAR => {
 890                 if !file_type.is_dir() {
 891                     bail!("got unexpected file type (expected directory)");
 892                 }
 893                 upload_list.push((BackupSpecificationType::PXAR, filename.to_owned(), format!("{}.didx", target), 0));
 894             }
 895             BackupSpecificationType::IMAGE => {
 896                 if !(file_type.is_file() || file_type.is_block_device()) {
 897                     bail!("got unexpected file type (expected file or block device)");
 898                 }
 899
 900                 let size = image_size(&PathBuf::from(filename))?;
 901
 902                 if size == 0 { bail!("got zero-sized file '{}'", filename); }
 903
 904                 upload_list.push((BackupSpecificationType::IMAGE, filename.to_owned(), format!("{}.fidx", target), size));
 905             }
 906             BackupSpecificationType::CONFIG => {
 907                 if !file_type.is_file() {
 908                     bail!("got unexpected file type (expected regular file)");
 909                 }
 910                 upload_list.push((BackupSpecificationType::CONFIG, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 911             }
 912             BackupSpecificationType::LOGFILE => {
 913                 if !file_type.is_file() {
 914                     bail!("got unexpected file type (expected regular file)");
 915                 }
 916                 upload_list.push((BackupSpecificationType::LOGFILE, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 917             }
 918         }
 919     }
 920
 921     let backup_time = Utc.timestamp(backup_time_opt.unwrap_or_else(|| Utc::now().timestamp()), 0);
 922
 923     let client = connect(repo.host(), repo.user())?;
 924     record_repository(&repo);
 925
 926     println!("Starting backup: {}/{}/{}", backup_type, backup_id, BackupDir::backup_time_to_string(backup_time));
 927
 928     println!("Client name: {}", proxmox::tools::nodename());
 929
 930     let start_time = Local::now();
 931
 932     println!("Starting protocol: {}", start_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, false));
 933
 934     let (crypt_config, rsa_encrypted_key) = match keyfile {
 935         None => (None, None),
 936         Some(path) => {
 937             let (key, created) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
 938
 939             let crypt_config = CryptConfig::new(key)?;
 940
 941             let path = master_pubkey_path()?;
 942             if path.exists() {
 943                 let pem_data = file_get_contents(&path)?;
 944                 let rsa = openssl::rsa::Rsa::public_key_from_pem(&pem_data)?;
 945                 let enc_key = crypt_config.generate_rsa_encoded_key(rsa, created)?;
 946                 (Some(Arc::new(crypt_config)), Some(enc_key))
 947             } else {
 948                 (Some(Arc::new(crypt_config)), None)
 949             }
 950         }
 951     };
 952
 953     let is_encrypted = Some(crypt_config.is_some());
 954
 955     let client = BackupWriter::start(
 956         client,
 957         repo.store(),
 958         backup_type,
 959         &backup_id,
 960         backup_time,
 961         verbose,
 962     ).await?;
 963
 964     let snapshot = BackupDir::new(backup_type, backup_id, backup_time.timestamp());
 965     let mut manifest = BackupManifest::new(snapshot);
 966
 967     let mut catalog = None;
 968     let mut catalog_result_tx = None;
 969
 970     for (backup_type, filename, target, size) in upload_list {
 971         match backup_type {
 972             BackupSpecificationType::CONFIG => {
 973                 println!("Upload config file '{}' to '{:?}' as {}", filename, repo, target);
 974                 let stats = client
 975                     .upload_blob_from_file(&filename, &target, crypt_config.clone(), true)
 976                     .await?;
 977                 manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
 978             }
 979             BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
 980                 println!("Upload log file '{}' to '{:?}' as {}", filename, repo, target);
 981                 let stats = client
 982                     .upload_blob_from_file(&filename, &target, crypt_config.clone(), true)
 983                     .await?;
 984                 manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
 985             }
 986             BackupSpecificationType::PXAR => {
 987                 // start catalog upload on first use
 988                 if catalog.is_none() {
 989                     let (cat, res) = spawn_catalog_upload(client.clone(), crypt_config.clone())?;
 990                     catalog = Some(cat);
 991                     catalog_result_tx = Some(res);
 992                 }
 993                 let catalog = catalog.as_ref().unwrap();
 994
 995                 println!("Upload directory '{}' to '{:?}' as {}", filename, repo, target);
 996                 catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
 997                 let stats = backup_directory(
 998                     &client,
 999                     &filename,
1000                     &target,
1001                     chunk_size_opt,
1002                     devices.clone(),
1003                     verbose,
1004                     skip_lost_and_found,
1005                     crypt_config.clone(),
1006                     catalog.clone(),
1007                     pattern_list.clone(),
1008                     entries_max as usize,
1009                 ).await?;
1010                 manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
1011                 catalog.lock().unwrap().end_directory()?;
1012             }
1013             BackupSpecificationType::IMAGE => {
1014                 println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
1015                 let stats = backup_image(
1016                     &client,
1017                     &filename,
1018                     &target,
1019                     size,
1020                     chunk_size_opt,
1021                     verbose,
1022                     crypt_config.clone(),
1023                 ).await?;
1024                 manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
1025             }
1026         }
1027     }
1028
1029     // finalize and upload catalog
1030     if let Some(catalog) = catalog {
1031         let mutex = Arc::try_unwrap(catalog)
1032             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
1033         let mut catalog = mutex.into_inner().unwrap();
1034
1035         catalog.finish()?;
1036
1037         drop(catalog); // close upload stream
1038
1039         if let Some(catalog_result_rx) = catalog_result_tx {
1040             let stats = catalog_result_rx.await??;
1041             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, is_encrypted)?;
1042         }
1043     }
1044
1045     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
1046         let target = "rsa-encrypted.key";
1047         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
1048         let stats = client
1049             .upload_blob_from_data(rsa_encrypted_key, target, None, false, false)
1050             .await?;
1051         manifest.add_file(format!("{}.blob", target), stats.size, stats.csum, is_encrypted)?;
1052
1053         // openssl rsautl -decrypt -inkey master-private.pem -in rsa-encrypted.key -out t
1054         /*
1055         let mut buffer2 = vec![0u8; rsa.size() as usize];
1056         let pem_data = file_get_contents("master-private.pem")?;
1057         let rsa = openssl::rsa::Rsa::private_key_from_pem(&pem_data)?;
1058         let len = rsa.private_decrypt(&buffer, &mut buffer2, openssl::rsa::Padding::PKCS1)?;
1059         println!("TEST {} {:?}", len, buffer2);
1060          */
1061     }
1062
1063     // create manifest (index.json)
1064     let manifest = manifest.into_json();
1065
1066     println!("Upload index.json to '{:?}'", repo);
1067     let manifest = serde_json::to_string_pretty(&manifest)?.into();
1068     client
1069         .upload_blob_from_data(manifest, MANIFEST_BLOB_NAME, crypt_config.clone(), true, true)
1070         .await?;
1071
1072     client.finish().await?;
1073
1074     let end_time = Local::now();
1075     let elapsed = end_time.signed_duration_since(start_time);
1076     println!("Duration: {}", elapsed);
1077
1078     println!("End Time: {}", end_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, false));
1079
1080     Ok(Value::Null)
1081 }
1082
1083 fn complete_backup_source(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1084
1085     let mut result = vec![];
1086
1087     let data: Vec<&str> = arg.splitn(2, ':').collect();
1088
1089     if data.len() != 2 {
1090         result.push(String::from("root.pxar:/"));
1091         result.push(String::from("etc.pxar:/etc"));
1092         return result;
1093     }
1094
1095     let files = tools::complete_file_name(data[1], param);
1096
1097     for file in files {
1098         result.push(format!("{}:{}", data[0], file));
1099     }
1100
1101     result
1102 }
1103
1104 async fn dump_image<W: Write>(
1105     client: Arc<BackupReader>,
1106     crypt_config: Option<Arc<CryptConfig>>,
1107     index: FixedIndexReader,
1108     mut writer: W,
1109     verbose: bool,
1110 ) -> Result<(), Error> {
1111
1112     let most_used = index.find_most_used_chunks(8);
1113
1114     let mut chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
1115
1116     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
1117     // and thus slows down reading. Instead, directly use RemoteChunkReader
1118     let mut per = 0;
1119     let mut bytes = 0;
1120     let start_time = std::time::Instant::now();
1121
1122     for pos in 0..index.index_count() {
1123         let digest = index.index_digest(pos).unwrap();
1124         let raw_data = chunk_reader.read_chunk(&digest).await?;
1125         writer.write_all(&raw_data)?;
1126         bytes += raw_data.len();
1127         if verbose {
1128             let next_per = ((pos+1)*100)/index.index_count();
1129             if per != next_per {
1130                 eprintln!("progress {}% (read {} bytes, duration {} sec)",
1131                           next_per, bytes, start_time.elapsed().as_secs());
1132                 per = next_per;
1133             }
1134         }
1135     }
1136
1137     let end_time = std::time::Instant::now();
1138     let elapsed = end_time.duration_since(start_time);
1139     eprintln!("restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1140               bytes,
1141               elapsed.as_secs_f64(),
1142               bytes as f64/(1024.0*1024.0*elapsed.as_secs_f64())
1143     );
1144
1145
1146     Ok(())
1147 }
1148
1149 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1150     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1151         (name.into(), archive_type(name).unwrap())
1152     } else if name.ends_with(".pxar") {
1153         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1154     } else if name.ends_with(".img") {
1155         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1156     } else {
1157         (format!("{}.blob", name), ArchiveType::Blob)
1158     }
1159 }
1160
1161 #[api(
1162    input: {
1163        properties: {
1164            repository: {
1165                schema: REPO_URL_SCHEMA,
1166                optional: true,
1167            },
1168            snapshot: {
1169                type: String,
1170                description: "Group/Snapshot path.",
1171            },
1172            "archive-name": {
1173                description: "Backup archive name.",
1174                type: String,
1175            },
1176            target: {
1177                type: String,
1178                description: r###"Target directory path. Use '-' to write to standard output.
1179
1180 We do not extraxt '.pxar' archives when writing to standard output.
1181
1182 "###
1183            },
1184            "allow-existing-dirs": {
1185                type: Boolean,
1186                description: "Do not fail if directories already exists.",
1187                optional: true,
1188            },
1189            keyfile: {
1190                schema: KEYFILE_SCHEMA,
1191                optional: true,
1192            },
1193        }
1194    }
1195 )]
1196 /// Restore backup repository.
1197 async fn restore(param: Value) -> Result<Value, Error> {
1198     let repo = extract_repository_from_value(&param)?;
1199
1200     let verbose = param["verbose"].as_bool().unwrap_or(false);
1201
1202     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1203
1204     let archive_name = tools::required_string_param(&param, "archive-name")?;
1205
1206     let client = connect(repo.host(), repo.user())?;
1207
1208     record_repository(&repo);
1209
1210     let path = tools::required_string_param(&param, "snapshot")?;
1211
1212     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
1213         let group = BackupGroup::parse(path)?;
1214         api_datastore_latest_snapshot(&client, repo.store(), group).await?
1215     } else {
1216         let snapshot = BackupDir::parse(path)?;
1217         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
1218     };
1219
1220     let target = tools::required_string_param(&param, "target")?;
1221     let target = if target == "-" { None } else { Some(target) };
1222
1223     let keyfile = param["keyfile"].as_str().map(PathBuf::from);
1224
1225     let crypt_config = match keyfile {
1226         None => None,
1227         Some(path) => {
1228             let (key, _) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
1229             Some(Arc::new(CryptConfig::new(key)?))
1230         }
1231     };
1232
1233     let client = BackupReader::start(
1234         client,
1235         crypt_config.clone(),
1236         repo.store(),
1237         &backup_type,
1238         &backup_id,
1239         backup_time,
1240         true,
1241     ).await?;
1242
1243     let manifest = client.download_manifest().await?;
1244
1245     let (archive_name, archive_type) = parse_archive_type(archive_name);
1246
1247     if archive_name == MANIFEST_BLOB_NAME {
1248         let backup_index_data = manifest.into_json().to_string();
1249         if let Some(target) = target {
1250             replace_file(target, backup_index_data.as_bytes(), CreateOptions::new())?;
1251         } else {
1252             let stdout = std::io::stdout();
1253             let mut writer = stdout.lock();
1254             writer.write_all(backup_index_data.as_bytes())
1255                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1256         }
1257
1258     } else if archive_type == ArchiveType::Blob {
1259
1260         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1261
1262         if let Some(target) = target {
1263            let mut writer = std::fs::OpenOptions::new()
1264                 .write(true)
1265                 .create(true)
1266                 .create_new(true)
1267                 .open(target)
1268                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?;
1269             std::io::copy(&mut reader, &mut writer)?;
1270         } else {
1271             let stdout = std::io::stdout();
1272             let mut writer = stdout.lock();
1273             std::io::copy(&mut reader, &mut writer)
1274                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1275         }
1276
1277     } else if archive_type == ArchiveType::DynamicIndex {
1278
1279         let index = client.download_dynamic_index(&manifest, &archive_name).await?;
1280
1281         let most_used = index.find_most_used_chunks(8);
1282
1283         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
1284
1285         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1286
1287         if let Some(target) = target {
1288             proxmox_backup::pxar::extract_archive(
1289                 pxar::decoder::Decoder::from_std(reader)?,
1290                 Path::new(target),
1291                 &[],
1292                 proxmox_backup::pxar::Flags::DEFAULT,
1293                 allow_existing_dirs,
1294                 |path| {
1295                     if verbose {
1296                         println!("{:?}", path);
1297                     }
1298                 },
1299             )
1300             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1301         } else {
1302             let mut writer = std::fs::OpenOptions::new()
1303                 .write(true)
1304                 .open("/dev/stdout")
1305                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1306
1307             std::io::copy(&mut reader, &mut writer)
1308                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1309         }
1310     } else if archive_type == ArchiveType::FixedIndex {
1311
1312         let index = client.download_fixed_index(&manifest, &archive_name).await?;
1313
1314         let mut writer = if let Some(target) = target {
1315             std::fs::OpenOptions::new()
1316                 .write(true)
1317                 .create(true)
1318                 .create_new(true)
1319                 .open(target)
1320                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1321         } else {
1322             std::fs::OpenOptions::new()
1323                 .write(true)
1324                 .open("/dev/stdout")
1325                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1326         };
1327
1328         dump_image(client.clone(), crypt_config.clone(), index, &mut writer, verbose).await?;
1329     }
1330
1331     Ok(Value::Null)
1332 }
1333
1334 #[api(
1335    input: {
1336        properties: {
1337            repository: {
1338                schema: REPO_URL_SCHEMA,
1339                optional: true,
1340            },
1341            snapshot: {
1342                type: String,
1343                description: "Group/Snapshot path.",
1344            },
1345            logfile: {
1346                type: String,
1347                description: "The path to the log file you want to upload.",
1348            },
1349            keyfile: {
1350                schema: KEYFILE_SCHEMA,
1351                optional: true,
1352            },
1353        }
1354    }
1355 )]
1356 /// Upload backup log file.
1357 async fn upload_log(param: Value) -> Result<Value, Error> {
1358
1359     let logfile = tools::required_string_param(&param, "logfile")?;
1360     let repo = extract_repository_from_value(&param)?;
1361
1362     let snapshot = tools::required_string_param(&param, "snapshot")?;
1363     let snapshot = BackupDir::parse(snapshot)?;
1364
1365     let mut client = connect(repo.host(), repo.user())?;
1366
1367     let keyfile = param["keyfile"].as_str().map(PathBuf::from);
1368
1369     let crypt_config = match keyfile {
1370         None => None,
1371         Some(path) => {
1372             let (key, _created) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
1373             let crypt_config = CryptConfig::new(key)?;
1374             Some(Arc::new(crypt_config))
1375         }
1376     };
1377
1378     let data = file_get_contents(logfile)?;
1379
1380     let blob = DataBlob::encode(&data, crypt_config.as_ref().map(Arc::as_ref), true)?;
1381
1382     let raw_data = blob.into_inner();
1383
1384     let path = format!("api2/json/admin/datastore/{}/upload-backup-log", repo.store());
1385
1386     let args = json!({
1387         "backup-type": snapshot.group().backup_type(),
1388         "backup-id":  snapshot.group().backup_id(),
1389         "backup-time": snapshot.backup_time().timestamp(),
1390     });
1391
1392     let body = hyper::Body::from(raw_data);
1393
1394     client.upload("application/octet-stream", body, &path, Some(args)).await
1395 }
1396
1397 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
1398     &ApiHandler::Async(&prune),
1399     &ObjectSchema::new(
1400         "Prune a backup repository.",
1401         &proxmox_backup::add_common_prune_prameters!([
1402             ("dry-run", true, &BooleanSchema::new(
1403                 "Just show what prune would do, but do not delete anything.")
1404              .schema()),
1405             ("group", false, &StringSchema::new("Backup group.").schema()),
1406         ], [
1407             ("output-format", true, &OUTPUT_FORMAT),
1408             (
1409                 "quiet",
1410                 true,
1411                 &BooleanSchema::new("Minimal output - only show removals.")
1412                     .schema()
1413             ),
1414             ("repository", true, &REPO_URL_SCHEMA),
1415         ])
1416     )
1417 );
1418
1419 fn prune<'a>(
1420     param: Value,
1421     _info: &ApiMethod,
1422     _rpcenv: &'a mut dyn RpcEnvironment,
1423 ) -> proxmox::api::ApiFuture<'a> {
1424     async move {
1425         prune_async(param).await
1426     }.boxed()
1427 }
1428
1429 async fn prune_async(mut param: Value) -> Result<Value, Error> {
1430     let repo = extract_repository_from_value(&param)?;
1431
1432     let mut client = connect(repo.host(), repo.user())?;
1433
1434     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1435
1436     let group = tools::required_string_param(&param, "group")?;
1437     let group = BackupGroup::parse(group)?;
1438
1439     let output_format = get_output_format(&param);
1440
1441     let quiet = param["quiet"].as_bool().unwrap_or(false);
1442
1443     param.as_object_mut().unwrap().remove("repository");
1444     param.as_object_mut().unwrap().remove("group");
1445     param.as_object_mut().unwrap().remove("output-format");
1446     param.as_object_mut().unwrap().remove("quiet");
1447
1448     param["backup-type"] = group.backup_type().into();
1449     param["backup-id"] = group.backup_id().into();
1450
1451     let mut result = client.post(&path, Some(param)).await?;
1452
1453     record_repository(&repo);
1454
1455     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1456         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1457         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time);
1458         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
1459     };
1460
1461     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1462         Ok(match v.as_bool() {
1463             Some(true) => "keep",
1464             Some(false) => "remove",
1465             None => "unknown",
1466         }.to_string())
1467     };
1468
1469     let options = default_table_format_options()
1470         .sortby("backup-type", false)
1471         .sortby("backup-id", false)
1472         .sortby("backup-time", false)
1473         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
1474         .column(ColumnConfig::new("backup-time").renderer(tools::format::render_epoch).header("date"))
1475         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
1476         ;
1477
1478     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_PRUNE;
1479
1480     let mut data = result["data"].take();
1481
1482     if quiet {
1483         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
1484             item["keep"].as_bool() == Some(false)
1485         }).map(|v| v.clone()).collect();
1486         data = list.into();
1487     }
1488
1489     format_and_print_result_full(&mut data, info, &output_format, &options);
1490
1491     Ok(Value::Null)
1492 }
1493
1494 #[api(
1495    input: {
1496        properties: {
1497            repository: {
1498                schema: REPO_URL_SCHEMA,
1499                optional: true,
1500            },
1501            "output-format": {
1502                schema: OUTPUT_FORMAT,
1503                optional: true,
1504            },
1505        }
1506    }
1507 )]
1508 /// Get repository status.
1509 async fn status(param: Value) -> Result<Value, Error> {
1510
1511     let repo = extract_repository_from_value(&param)?;
1512
1513     let output_format = get_output_format(&param);
1514
1515     let client = connect(repo.host(), repo.user())?;
1516
1517     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1518
1519     let mut result = client.get(&path, None).await?;
1520     let mut data = result["data"].take();
1521
1522     record_repository(&repo);
1523
1524     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1525         let v = v.as_u64().unwrap();
1526         let total = record["total"].as_u64().unwrap();
1527         let roundup = total/200;
1528         let per = ((v+roundup)*100)/total;
1529         let info = format!(" ({} %)", per);
1530         Ok(format!("{} {:>8}", v, info))
1531     };
1532
1533     let options = default_table_format_options()
1534         .noheader(true)
1535         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1536         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1537         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1538
1539     let schema = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_STATUS;
1540
1541     format_and_print_result_full(&mut data, schema, &output_format, &options);
1542
1543     Ok(Value::Null)
1544 }
1545
1546 // like get, but simply ignore errors and return Null instead
1547 async fn try_get(repo: &BackupRepository, url: &str) -> Value {
1548
1549     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
1550     let password = std::env::var(ENV_VAR_PBS_PASSWORD).ok();
1551
1552     let options = HttpClientOptions::new()
1553         .prefix(Some("proxmox-backup".to_string()))
1554         .password(password)
1555         .interactive(false)
1556         .fingerprint(fingerprint)
1557         .fingerprint_cache(true)
1558         .ticket_cache(true);
1559
1560     let client = match HttpClient::new(repo.host(), repo.user(), options) {
1561         Ok(v) => v,
1562         _ => return Value::Null,
1563     };
1564
1565     let mut resp = match client.get(url, None).await {
1566         Ok(v) => v,
1567         _ => return Value::Null,
1568     };
1569
1570     if let Some(map) = resp.as_object_mut() {
1571         if let Some(data) = map.remove("data") {
1572             return data;
1573         }
1574     }
1575     Value::Null
1576 }
1577
1578 fn complete_backup_group(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1579     proxmox_backup::tools::runtime::main(async { complete_backup_group_do(param).await })
1580 }
1581
1582 async fn complete_backup_group_do(param: &HashMap<String, String>) -> Vec<String> {
1583
1584     let mut result = vec![];
1585
1586     let repo = match extract_repository_from_map(param) {
1587         Some(v) => v,
1588         _ => return result,
1589     };
1590
1591     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
1592
1593     let data = try_get(&repo, &path).await;
1594
1595     if let Some(list) = data.as_array() {
1596         for item in list {
1597             if let (Some(backup_id), Some(backup_type)) =
1598                 (item["backup-id"].as_str(), item["backup-type"].as_str())
1599             {
1600                 result.push(format!("{}/{}", backup_type, backup_id));
1601             }
1602         }
1603     }
1604
1605     result
1606 }
1607
1608 fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1609     proxmox_backup::tools::runtime::main(async { complete_group_or_snapshot_do(arg, param).await })
1610 }
1611
1612 async fn complete_group_or_snapshot_do(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1613
1614     if arg.matches('/').count() < 2 {
1615         let groups = complete_backup_group_do(param).await;
1616         let mut result = vec![];
1617         for group in groups {
1618             result.push(group.to_string());
1619             result.push(format!("{}/", group));
1620         }
1621         return result;
1622     }
1623
1624     complete_backup_snapshot_do(param).await
1625 }
1626
1627 fn complete_backup_snapshot(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1628     proxmox_backup::tools::runtime::main(async { complete_backup_snapshot_do(param).await })
1629 }
1630
1631 async fn complete_backup_snapshot_do(param: &HashMap<String, String>) -> Vec<String> {
1632
1633     let mut result = vec![];
1634
1635     let repo = match extract_repository_from_map(param) {
1636         Some(v) => v,
1637         _ => return result,
1638     };
1639
1640     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
1641
1642     let data = try_get(&repo, &path).await;
1643
1644     if let Some(list) = data.as_array() {
1645         for item in list {
1646             if let (Some(backup_id), Some(backup_type), Some(backup_time)) =
1647                 (item["backup-id"].as_str(), item["backup-type"].as_str(), item["backup-time"].as_i64())
1648             {
1649                 let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
1650                 result.push(snapshot.relative_path().to_str().unwrap().to_owned());
1651             }
1652         }
1653     }
1654
1655     result
1656 }
1657
1658 fn complete_server_file_name(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1659     proxmox_backup::tools::runtime::main(async { complete_server_file_name_do(param).await })
1660 }
1661
1662 async fn complete_server_file_name_do(param: &HashMap<String, String>) -> Vec<String> {
1663
1664     let mut result = vec![];
1665
1666     let repo = match extract_repository_from_map(param) {
1667         Some(v) => v,
1668         _ => return result,
1669     };
1670
1671     let snapshot = match param.get("snapshot") {
1672         Some(path) => {
1673             match BackupDir::parse(path) {
1674                 Ok(v) => v,
1675                 _ => return result,
1676             }
1677         }
1678         _ => return result,
1679     };
1680
1681     let query = tools::json_object_to_query(json!({
1682         "backup-type": snapshot.group().backup_type(),
1683         "backup-id": snapshot.group().backup_id(),
1684         "backup-time": snapshot.backup_time().timestamp(),
1685     })).unwrap();
1686
1687     let path = format!("api2/json/admin/datastore/{}/files?{}", repo.store(), query);
1688
1689     let data = try_get(&repo, &path).await;
1690
1691     if let Some(list) = data.as_array() {
1692         for item in list {
1693             if let Some(filename) = item["filename"].as_str() {
1694                 result.push(filename.to_owned());
1695             }
1696         }
1697     }
1698
1699     result
1700 }
1701
1702 fn complete_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1703     complete_server_file_name(arg, param)
1704         .iter()
1705         .map(|v| tools::format::strip_server_file_expenstion(&v))
1706         .collect()
1707 }
1708
1709 fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1710     complete_server_file_name(arg, param)
1711         .iter()
1712         .filter_map(|v| {
1713             let name = tools::format::strip_server_file_expenstion(&v);
1714             if name.ends_with(".pxar") {
1715                 Some(name)
1716             } else {
1717                 None
1718             }
1719         })
1720         .collect()
1721 }
1722
1723 fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
1724
1725     let mut result = vec![];
1726
1727     let mut size = 64;
1728     loop {
1729         result.push(size.to_string());
1730         size *= 2;
1731         if size > 4096 { break; }
1732     }
1733
1734     result
1735 }
1736
1737 fn get_encryption_key_password() -> Result<Vec<u8>, Error> {
1738
1739     // fixme: implement other input methods
1740
1741     use std::env::VarError::*;
1742     match std::env::var("PBS_ENCRYPTION_PASSWORD") {
1743         Ok(p) => return Ok(p.as_bytes().to_vec()),
1744         Err(NotUnicode(_)) => bail!("PBS_ENCRYPTION_PASSWORD contains bad characters"),
1745         Err(NotPresent) => {
1746             // Try another method
1747         }
1748     }
1749
1750     // If we're on a TTY, query the user for a password
1751     if tty::stdin_isatty() {
1752         return Ok(tty::read_password("Encryption Key Password: ")?);
1753     }
1754
1755     bail!("no password input mechanism available");
1756 }
1757
1758 fn key_create(
1759     param: Value,
1760     _info: &ApiMethod,
1761     _rpcenv: &mut dyn RpcEnvironment,
1762 ) -> Result<Value, Error> {
1763
1764     let path = tools::required_string_param(&param, "path")?;
1765     let path = PathBuf::from(path);
1766
1767     let kdf = param["kdf"].as_str().unwrap_or("scrypt");
1768
1769     let key = proxmox::sys::linux::random_data(32)?;
1770
1771     if kdf == "scrypt" {
1772         // always read passphrase from tty
1773         if !tty::stdin_isatty() {
1774             bail!("unable to read passphrase - no tty");
1775         }
1776
1777         let password = tty::read_and_verify_password("Encryption Key Password: ")?;
1778
1779         let key_config = encrypt_key_with_passphrase(&key, &password)?;
1780
1781         store_key_config(&path, false, key_config)?;
1782
1783         Ok(Value::Null)
1784     } else if kdf == "none" {
1785         let created =  Local.timestamp(Local::now().timestamp(), 0);
1786
1787         store_key_config(&path, false, KeyConfig {
1788             kdf: None,
1789             created,
1790             modified: created,
1791             data: key,
1792         })?;
1793
1794         Ok(Value::Null)
1795     } else {
1796         unreachable!();
1797     }
1798 }
1799
1800 fn master_pubkey_path() -> Result<PathBuf, Error> {
1801     let base = BaseDirectories::with_prefix("proxmox-backup")?;
1802
1803     // usually $HOME/.config/proxmox-backup/master-public.pem
1804     let path = base.place_config_file("master-public.pem")?;
1805
1806     Ok(path)
1807 }
1808
1809 fn key_import_master_pubkey(
1810     param: Value,
1811     _info: &ApiMethod,
1812     _rpcenv: &mut dyn RpcEnvironment,
1813 ) -> Result<Value, Error> {
1814
1815     let path = tools::required_string_param(&param, "path")?;
1816     let path = PathBuf::from(path);
1817
1818     let pem_data = file_get_contents(&path)?;
1819
1820     if let Err(err) = openssl::pkey::PKey::public_key_from_pem(&pem_data) {
1821         bail!("Unable to decode PEM data - {}", err);
1822     }
1823
1824     let target_path = master_pubkey_path()?;
1825
1826     replace_file(&target_path, &pem_data, CreateOptions::new())?;
1827
1828     println!("Imported public master key to {:?}", target_path);
1829
1830     Ok(Value::Null)
1831 }
1832
1833 fn key_create_master_key(
1834     _param: Value,
1835     _info: &ApiMethod,
1836     _rpcenv: &mut dyn RpcEnvironment,
1837 ) -> Result<Value, Error> {
1838
1839     // we need a TTY to query the new password
1840     if !tty::stdin_isatty() {
1841         bail!("unable to create master key - no tty");
1842     }
1843
1844     let rsa = openssl::rsa::Rsa::generate(4096)?;
1845     let pkey = openssl::pkey::PKey::from_rsa(rsa)?;
1846
1847
1848     let password = String::from_utf8(tty::read_and_verify_password("Master Key Password: ")?)?;
1849
1850     let pub_key: Vec<u8> = pkey.public_key_to_pem()?;
1851     let filename_pub = "master-public.pem";
1852     println!("Writing public master key to {}", filename_pub);
1853     replace_file(filename_pub, pub_key.as_slice(), CreateOptions::new())?;
1854
1855     let cipher = openssl::symm::Cipher::aes_256_cbc();
1856     let priv_key: Vec<u8> = pkey.private_key_to_pem_pkcs8_passphrase(cipher, password.as_bytes())?;
1857
1858     let filename_priv = "master-private.pem";
1859     println!("Writing private master key to {}", filename_priv);
1860     replace_file(filename_priv, priv_key.as_slice(), CreateOptions::new())?;
1861
1862     Ok(Value::Null)
1863 }
1864
1865 fn key_change_passphrase(
1866     param: Value,
1867     _info: &ApiMethod,
1868     _rpcenv: &mut dyn RpcEnvironment,
1869 ) -> Result<Value, Error> {
1870
1871     let path = tools::required_string_param(&param, "path")?;
1872     let path = PathBuf::from(path);
1873
1874     let kdf = param["kdf"].as_str().unwrap_or("scrypt");
1875
1876     // we need a TTY to query the new password
1877     if !tty::stdin_isatty() {
1878         bail!("unable to change passphrase - no tty");
1879     }
1880
1881     let (key, created) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
1882
1883     if kdf == "scrypt" {
1884
1885         let password = tty::read_and_verify_password("New Password: ")?;
1886
1887         let mut new_key_config = encrypt_key_with_passphrase(&key, &password)?;
1888         new_key_config.created = created; // keep original value
1889
1890         store_key_config(&path, true, new_key_config)?;
1891
1892         Ok(Value::Null)
1893     } else if kdf == "none" {
1894         let modified =  Local.timestamp(Local::now().timestamp(), 0);
1895
1896         store_key_config(&path, true, KeyConfig {
1897             kdf: None,
1898             created, // keep original value
1899             modified,
1900             data: key.to_vec(),
1901         })?;
1902
1903         Ok(Value::Null)
1904     } else {
1905         unreachable!();
1906     }
1907 }
1908
1909 fn key_mgmt_cli() -> CliCommandMap {
1910
1911     const KDF_SCHEMA: Schema =
1912         StringSchema::new("Key derivation function. Choose 'none' to store the key unecrypted.")
1913         .format(&ApiStringFormat::Enum(&[
1914             EnumEntry::new("scrypt", "SCrypt"),
1915             EnumEntry::new("none", "Do not encrypt the key")]))
1916         .default("scrypt")
1917         .schema();
1918
1919     #[sortable]
1920     const API_METHOD_KEY_CREATE: ApiMethod = ApiMethod::new(
1921         &ApiHandler::Sync(&key_create),
1922         &ObjectSchema::new(
1923             "Create a new encryption key.",
1924             &sorted!([
1925                 ("path", false, &StringSchema::new("File system path.").schema()),
1926                 ("kdf", true, &KDF_SCHEMA),
1927             ]),
1928         )
1929     );
1930
1931     let key_create_cmd_def = CliCommand::new(&API_METHOD_KEY_CREATE)
1932         .arg_param(&["path"])
1933         .completion_cb("path", tools::complete_file_name);
1934
1935     #[sortable]
1936     const API_METHOD_KEY_CHANGE_PASSPHRASE: ApiMethod = ApiMethod::new(
1937         &ApiHandler::Sync(&key_change_passphrase),
1938         &ObjectSchema::new(
1939             "Change the passphrase required to decrypt the key.",
1940             &sorted!([
1941                 ("path", false, &StringSchema::new("File system path.").schema()),
1942                 ("kdf", true, &KDF_SCHEMA),
1943             ]),
1944         )
1945     );
1946
1947     let key_change_passphrase_cmd_def = CliCommand::new(&API_METHOD_KEY_CHANGE_PASSPHRASE)
1948         .arg_param(&["path"])
1949         .completion_cb("path", tools::complete_file_name);
1950
1951     const API_METHOD_KEY_CREATE_MASTER_KEY: ApiMethod = ApiMethod::new(
1952         &ApiHandler::Sync(&key_create_master_key),
1953         &ObjectSchema::new("Create a new 4096 bit RSA master pub/priv key pair.", &[])
1954     );
1955
1956     let key_create_master_key_cmd_def = CliCommand::new(&API_METHOD_KEY_CREATE_MASTER_KEY);
1957
1958     #[sortable]
1959     const API_METHOD_KEY_IMPORT_MASTER_PUBKEY: ApiMethod = ApiMethod::new(
1960         &ApiHandler::Sync(&key_import_master_pubkey),
1961         &ObjectSchema::new(
1962             "Import a new RSA public key and use it as master key. The key is expected to be in '.pem' format.",
1963             &sorted!([ ("path", false, &StringSchema::new("File system path.").schema()) ]),
1964         )
1965     );
1966
1967     let key_import_master_pubkey_cmd_def = CliCommand::new(&API_METHOD_KEY_IMPORT_MASTER_PUBKEY)
1968         .arg_param(&["path"])
1969         .completion_cb("path", tools::complete_file_name);
1970
1971     CliCommandMap::new()
1972         .insert("create", key_create_cmd_def)
1973         .insert("create-master-key", key_create_master_key_cmd_def)
1974         .insert("import-master-pubkey", key_import_master_pubkey_cmd_def)
1975         .insert("change-passphrase", key_change_passphrase_cmd_def)
1976 }
1977
1978 fn mount(
1979     param: Value,
1980     _info: &ApiMethod,
1981     _rpcenv: &mut dyn RpcEnvironment,
1982 ) -> Result<Value, Error> {
1983     let verbose = param["verbose"].as_bool().unwrap_or(false);
1984     if verbose {
1985         // This will stay in foreground with debug output enabled as None is
1986         // passed for the RawFd.
1987         return proxmox_backup::tools::runtime::main(mount_do(param, None));
1988     }
1989
1990     // Process should be deamonized.
1991     // Make sure to fork before the async runtime is instantiated to avoid troubles.
1992     let pipe = pipe()?;
1993     match fork() {
1994         Ok(ForkResult::Parent { .. }) => {
1995             nix::unistd::close(pipe.1).unwrap();
1996             // Blocks the parent process until we are ready to go in the child
1997             let _res = nix::unistd::read(pipe.0, &mut [0]).unwrap();
1998             Ok(Value::Null)
1999         }
2000         Ok(ForkResult::Child) => {
2001             nix::unistd::close(pipe.0).unwrap();
2002             nix::unistd::setsid().unwrap();
2003             proxmox_backup::tools::runtime::main(mount_do(param, Some(pipe.1)))
2004         }
2005         Err(_) => bail!("failed to daemonize process"),
2006     }
2007 }
2008
2009 use proxmox_backup::client::RemoteChunkReader;
2010 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
2011 /// async use!
2012 ///
2013 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
2014 /// so that we can properly access it from multiple threads simultaneously while not issuing
2015 /// duplicate simultaneous reads over http.
2016 struct BufferedDynamicReadAt {
2017     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
2018 }
2019
2020 impl BufferedDynamicReadAt {
2021     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
2022         Self {
2023             inner: Mutex::new(inner),
2024         }
2025     }
2026 }
2027
2028 impl pxar::accessor::ReadAt for BufferedDynamicReadAt {
2029     fn poll_read_at(
2030         self: Pin<&Self>,
2031         _cx: &mut Context,
2032         buf: &mut [u8],
2033         offset: u64,
2034     ) -> Poll<io::Result<usize>> {
2035         use std::io::Read;
2036         tokio::task::block_in_place(move || {
2037             let mut reader = self.inner.lock().unwrap();
2038             reader.seek(SeekFrom::Start(offset))?;
2039             Poll::Ready(Ok(reader.read(buf)?))
2040         })
2041     }
2042 }
2043
2044 async fn mount_do(param: Value, pipe: Option<RawFd>) -> Result<Value, Error> {
2045     let repo = extract_repository_from_value(&param)?;
2046     let archive_name = tools::required_string_param(&param, "archive-name")?;
2047     let target = tools::required_string_param(&param, "target")?;
2048     let client = connect(repo.host(), repo.user())?;
2049
2050     record_repository(&repo);
2051
2052     let path = tools::required_string_param(&param, "snapshot")?;
2053     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
2054         let group = BackupGroup::parse(path)?;
2055         api_datastore_latest_snapshot(&client, repo.store(), group).await?
2056     } else {
2057         let snapshot = BackupDir::parse(path)?;
2058         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
2059     };
2060
2061     let keyfile = param["keyfile"].as_str().map(PathBuf::from);
2062     let crypt_config = match keyfile {
2063         None => None,
2064         Some(path) => {
2065             let (key, _) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
2066             Some(Arc::new(CryptConfig::new(key)?))
2067         }
2068     };
2069
2070     let server_archive_name = if archive_name.ends_with(".pxar") {
2071         format!("{}.didx", archive_name)
2072     } else {
2073         bail!("Can only mount pxar archives.");
2074     };
2075
2076     let client = BackupReader::start(
2077         client,
2078         crypt_config.clone(),
2079         repo.store(),
2080         &backup_type,
2081         &backup_id,
2082         backup_time,
2083         true,
2084     ).await?;
2085
2086     let manifest = client.download_manifest().await?;
2087
2088     if server_archive_name.ends_with(".didx") {
2089         let index = client.download_dynamic_index(&manifest, &server_archive_name).await?;
2090         let most_used = index.find_most_used_chunks(8);
2091         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
2092         let reader = BufferedDynamicReader::new(index, chunk_reader);
2093         let archive_size = reader.archive_size();
2094         let reader: proxmox_backup::pxar::fuse::Reader =
2095             Arc::new(BufferedDynamicReadAt::new(reader));
2096         let decoder = proxmox_backup::pxar::fuse::Accessor::new(reader, archive_size).await?;
2097         let options = OsStr::new("ro,default_permissions");
2098
2099         let session = proxmox_backup::pxar::fuse::Session::mount(
2100             decoder,
2101             &options,
2102             false,
2103             Path::new(target),
2104         )
2105         .map_err(|err| format_err!("pxar mount failed: {}", err))?;
2106
2107         if let Some(pipe) = pipe {
2108             nix::unistd::chdir(Path::new("/")).unwrap();
2109             // Finish creation of daemon by redirecting filedescriptors.
2110             let nullfd = nix::fcntl::open(
2111                 "/dev/null",
2112                 nix::fcntl::OFlag::O_RDWR,
2113                 nix::sys::stat::Mode::empty(),
2114             ).unwrap();
2115             nix::unistd::dup2(nullfd, 0).unwrap();
2116             nix::unistd::dup2(nullfd, 1).unwrap();
2117             nix::unistd::dup2(nullfd, 2).unwrap();
2118             if nullfd > 2 {
2119                 nix::unistd::close(nullfd).unwrap();
2120             }
2121             // Signal the parent process that we are done with the setup and it can
2122             // terminate.
2123             nix::unistd::write(pipe, &[0u8])?;
2124             nix::unistd::close(pipe).unwrap();
2125         }
2126
2127         let mut interrupt = signal(SignalKind::interrupt())?;
2128         select! {
2129             res = session.fuse() => res?,
2130             _ = interrupt.recv().fuse() => {
2131                 // exit on interrupted
2132             }
2133         }
2134     } else {
2135         bail!("unknown archive file extension (expected .pxar)");
2136     }
2137
2138     Ok(Value::Null)
2139 }
2140
2141 #[api(
2142     input: {
2143         properties: {
2144             "snapshot": {
2145                 type: String,
2146                 description: "Group/Snapshot path.",
2147             },
2148             "archive-name": {
2149                 type: String,
2150                 description: "Backup archive name.",
2151             },
2152             "repository": {
2153                 optional: true,
2154                 schema: REPO_URL_SCHEMA,
2155             },
2156             "keyfile": {
2157                 optional: true,
2158                 type: String,
2159                 description: "Path to encryption key.",
2160             },
2161         },
2162     },
2163 )]
2164 /// Shell to interactively inspect and restore snapshots.
2165 async fn catalog_shell(param: Value) -> Result<(), Error> {
2166     let repo = extract_repository_from_value(&param)?;
2167     let client = connect(repo.host(), repo.user())?;
2168     let path = tools::required_string_param(&param, "snapshot")?;
2169     let archive_name = tools::required_string_param(&param, "archive-name")?;
2170
2171     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
2172         let group = BackupGroup::parse(path)?;
2173         api_datastore_latest_snapshot(&client, repo.store(), group).await?
2174     } else {
2175         let snapshot = BackupDir::parse(path)?;
2176         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
2177     };
2178
2179     let keyfile = param["keyfile"].as_str().map(|p| PathBuf::from(p));
2180     let crypt_config = match keyfile {
2181         None => None,
2182         Some(path) => {
2183             let (key, _) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
2184             Some(Arc::new(CryptConfig::new(key)?))
2185         }
2186     };
2187
2188     let server_archive_name = if archive_name.ends_with(".pxar") {
2189         format!("{}.didx", archive_name)
2190     } else {
2191         bail!("Can only mount pxar archives.");
2192     };
2193
2194     let client = BackupReader::start(
2195         client,
2196         crypt_config.clone(),
2197         repo.store(),
2198         &backup_type,
2199         &backup_id,
2200         backup_time,
2201         true,
2202     ).await?;
2203
2204     let mut tmpfile = std::fs::OpenOptions::new()
2205         .write(true)
2206         .read(true)
2207         .custom_flags(libc::O_TMPFILE)
2208         .open("/tmp")?;
2209
2210     let manifest = client.download_manifest().await?;
2211
2212     let index = client.download_dynamic_index(&manifest, &server_archive_name).await?;
2213     let most_used = index.find_most_used_chunks(8);
2214     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config.clone(), most_used);
2215     let reader = BufferedDynamicReader::new(index, chunk_reader);
2216     let archive_size = reader.archive_size();
2217     let reader: proxmox_backup::pxar::fuse::Reader =
2218         Arc::new(BufferedDynamicReadAt::new(reader));
2219     let decoder = proxmox_backup::pxar::fuse::Accessor::new(reader, archive_size).await?;
2220
2221     client.download(CATALOG_NAME, &mut tmpfile).await?;
2222     let index = DynamicIndexReader::new(tmpfile)
2223         .map_err(|err| format_err!("unable to read catalog index - {}", err))?;
2224
2225     // Note: do not use values stored in index (not trusted) - instead, computed them again
2226     let (csum, size) = index.compute_csum();
2227     manifest.verify_file(CATALOG_NAME, &csum, size)?;
2228
2229     let most_used = index.find_most_used_chunks(8);
2230     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
2231     let mut reader = BufferedDynamicReader::new(index, chunk_reader);
2232     let mut catalogfile = std::fs::OpenOptions::new()
2233         .write(true)
2234         .read(true)
2235         .custom_flags(libc::O_TMPFILE)
2236         .open("/tmp")?;
2237
2238     std::io::copy(&mut reader, &mut catalogfile)
2239         .map_err(|err| format_err!("unable to download catalog - {}", err))?;
2240
2241     catalogfile.seek(SeekFrom::Start(0))?;
2242     let catalog_reader = CatalogReader::new(catalogfile);
2243     let state = Shell::new(
2244         catalog_reader,
2245         &server_archive_name,
2246         decoder,
2247     ).await?;
2248
2249     println!("Starting interactive shell");
2250     state.shell().await?;
2251
2252     record_repository(&repo);
2253
2254     Ok(())
2255 }
2256
2257 fn catalog_mgmt_cli() -> CliCommandMap {
2258     let catalog_shell_cmd_def = CliCommand::new(&API_METHOD_CATALOG_SHELL)
2259         .arg_param(&["snapshot", "archive-name"])
2260         .completion_cb("repository", complete_repository)
2261         .completion_cb("archive-name", complete_pxar_archive_name)
2262         .completion_cb("snapshot", complete_group_or_snapshot);
2263
2264     let catalog_dump_cmd_def = CliCommand::new(&API_METHOD_DUMP_CATALOG)
2265         .arg_param(&["snapshot"])
2266         .completion_cb("repository", complete_repository)
2267         .completion_cb("snapshot", complete_backup_snapshot);
2268
2269     CliCommandMap::new()
2270         .insert("dump", catalog_dump_cmd_def)
2271         .insert("shell", catalog_shell_cmd_def)
2272 }
2273
2274 #[api(
2275     input: {
2276         properties: {
2277             repository: {
2278                 schema: REPO_URL_SCHEMA,
2279                 optional: true,
2280             },
2281             limit: {
2282                 description: "The maximal number of tasks to list.",
2283                 type: Integer,
2284                 optional: true,
2285                 minimum: 1,
2286                 maximum: 1000,
2287                 default: 50,
2288             },
2289             "output-format": {
2290                 schema: OUTPUT_FORMAT,
2291                 optional: true,
2292             },
2293             all: {
2294                 type: Boolean,
2295                 description: "Also list stopped tasks.",
2296                 optional: true,
2297             },
2298         }
2299     }
2300 )]
2301 /// List running server tasks for this repo user
2302 async fn task_list(param: Value) -> Result<Value, Error> {
2303
2304     let output_format = get_output_format(&param);
2305
2306     let repo = extract_repository_from_value(&param)?;
2307     let client = connect(repo.host(), repo.user())?;
2308
2309     let limit = param["limit"].as_u64().unwrap_or(50) as usize;
2310     let running = !param["all"].as_bool().unwrap_or(false);
2311
2312     let args = json!({
2313         "running": running,
2314         "start": 0,
2315         "limit": limit,
2316         "userfilter": repo.user(),
2317         "store": repo.store(),
2318     });
2319
2320     let mut result = client.get("api2/json/nodes/localhost/tasks", Some(args)).await?;
2321     let mut data = result["data"].take();
2322
2323     let schema = &proxmox_backup::api2::node::tasks::API_RETURN_SCHEMA_LIST_TASKS;
2324
2325     let options = default_table_format_options()
2326         .column(ColumnConfig::new("starttime").right_align(false).renderer(tools::format::render_epoch))
2327         .column(ColumnConfig::new("endtime").right_align(false).renderer(tools::format::render_epoch))
2328         .column(ColumnConfig::new("upid"))
2329         .column(ColumnConfig::new("status").renderer(tools::format::render_task_status));
2330
2331     format_and_print_result_full(&mut data, schema, &output_format, &options);
2332
2333     Ok(Value::Null)
2334 }
2335
2336 #[api(
2337     input: {
2338         properties: {
2339             repository: {
2340                 schema: REPO_URL_SCHEMA,
2341                 optional: true,
2342             },
2343             upid: {
2344                 schema: UPID_SCHEMA,
2345             },
2346         }
2347     }
2348 )]
2349 /// Display the task log.
2350 async fn task_log(param: Value) -> Result<Value, Error> {
2351
2352     let repo = extract_repository_from_value(&param)?;
2353     let upid =  tools::required_string_param(&param, "upid")?;
2354
2355     let client = connect(repo.host(), repo.user())?;
2356
2357     display_task_log(client, upid, true).await?;
2358
2359     Ok(Value::Null)
2360 }
2361
2362 #[api(
2363     input: {
2364         properties: {
2365             repository: {
2366                 schema: REPO_URL_SCHEMA,
2367                 optional: true,
2368             },
2369             upid: {
2370                 schema: UPID_SCHEMA,
2371             },
2372         }
2373     }
2374 )]
2375 /// Try to stop a specific task.
2376 async fn task_stop(param: Value) -> Result<Value, Error> {
2377
2378     let repo = extract_repository_from_value(&param)?;
2379     let upid_str =  tools::required_string_param(&param, "upid")?;
2380
2381     let mut client = connect(repo.host(), repo.user())?;
2382
2383     let path = format!("api2/json/nodes/localhost/tasks/{}", upid_str);
2384     let _ = client.delete(&path, None).await?;
2385
2386     Ok(Value::Null)
2387 }
2388
2389 fn task_mgmt_cli() -> CliCommandMap {
2390
2391     let task_list_cmd_def = CliCommand::new(&API_METHOD_TASK_LIST)
2392         .completion_cb("repository", complete_repository);
2393
2394     let task_log_cmd_def = CliCommand::new(&API_METHOD_TASK_LOG)
2395         .arg_param(&["upid"]);
2396
2397     let task_stop_cmd_def = CliCommand::new(&API_METHOD_TASK_STOP)
2398         .arg_param(&["upid"]);
2399
2400     CliCommandMap::new()
2401         .insert("log", task_log_cmd_def)
2402         .insert("list", task_list_cmd_def)
2403         .insert("stop", task_stop_cmd_def)
2404 }
2405
2406 fn main() {
2407
2408     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
2409         .arg_param(&["backupspec"])
2410         .completion_cb("repository", complete_repository)
2411         .completion_cb("backupspec", complete_backup_source)
2412         .completion_cb("keyfile", tools::complete_file_name)
2413         .completion_cb("chunk-size", complete_chunk_size);
2414
2415     let upload_log_cmd_def = CliCommand::new(&API_METHOD_UPLOAD_LOG)
2416         .arg_param(&["snapshot", "logfile"])
2417         .completion_cb("snapshot", complete_backup_snapshot)
2418         .completion_cb("logfile", tools::complete_file_name)
2419         .completion_cb("keyfile", tools::complete_file_name)
2420         .completion_cb("repository", complete_repository);
2421
2422     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
2423         .completion_cb("repository", complete_repository);
2424
2425     let snapshots_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOTS)
2426         .arg_param(&["group"])
2427         .completion_cb("group", complete_backup_group)
2428         .completion_cb("repository", complete_repository);
2429
2430     let forget_cmd_def = CliCommand::new(&API_METHOD_FORGET_SNAPSHOTS)
2431         .arg_param(&["snapshot"])
2432         .completion_cb("repository", complete_repository)
2433         .completion_cb("snapshot", complete_backup_snapshot);
2434
2435     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
2436         .completion_cb("repository", complete_repository);
2437
2438     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
2439         .arg_param(&["snapshot", "archive-name", "target"])
2440         .completion_cb("repository", complete_repository)
2441         .completion_cb("snapshot", complete_group_or_snapshot)
2442         .completion_cb("archive-name", complete_archive_name)
2443         .completion_cb("target", tools::complete_file_name);
2444
2445     let files_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOT_FILES)
2446         .arg_param(&["snapshot"])
2447         .completion_cb("repository", complete_repository)
2448         .completion_cb("snapshot", complete_backup_snapshot);
2449
2450     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
2451         .arg_param(&["group"])
2452         .completion_cb("group", complete_backup_group)
2453         .completion_cb("repository", complete_repository);
2454
2455     let status_cmd_def = CliCommand::new(&API_METHOD_STATUS)
2456         .completion_cb("repository", complete_repository);
2457
2458     let login_cmd_def = CliCommand::new(&API_METHOD_API_LOGIN)
2459         .completion_cb("repository", complete_repository);
2460
2461     let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
2462         .completion_cb("repository", complete_repository);
2463
2464     #[sortable]
2465     const API_METHOD_MOUNT: ApiMethod = ApiMethod::new(
2466         &ApiHandler::Sync(&mount),
2467         &ObjectSchema::new(
2468             "Mount pxar archive.",
2469             &sorted!([
2470                 ("snapshot", false, &StringSchema::new("Group/Snapshot path.").schema()),
2471                 ("archive-name", false, &StringSchema::new("Backup archive name.").schema()),
2472                 ("target", false, &StringSchema::new("Target directory path.").schema()),
2473                 ("repository", true, &REPO_URL_SCHEMA),
2474                 ("keyfile", true, &StringSchema::new("Path to encryption key.").schema()),
2475                 ("verbose", true, &BooleanSchema::new("Verbose output.").default(false).schema()),
2476             ]),
2477         )
2478     );
2479
2480     let mount_cmd_def = CliCommand::new(&API_METHOD_MOUNT)
2481         .arg_param(&["snapshot", "archive-name", "target"])
2482         .completion_cb("repository", complete_repository)
2483         .completion_cb("snapshot", complete_group_or_snapshot)
2484         .completion_cb("archive-name", complete_pxar_archive_name)
2485         .completion_cb("target", tools::complete_file_name);
2486
2487
2488     let cmd_def = CliCommandMap::new()
2489         .insert("backup", backup_cmd_def)
2490         .insert("upload-log", upload_log_cmd_def)
2491         .insert("forget", forget_cmd_def)
2492         .insert("garbage-collect", garbage_collect_cmd_def)
2493         .insert("list", list_cmd_def)
2494         .insert("login", login_cmd_def)
2495         .insert("logout", logout_cmd_def)
2496         .insert("prune", prune_cmd_def)
2497         .insert("restore", restore_cmd_def)
2498         .insert("snapshots", snapshots_cmd_def)
2499         .insert("files", files_cmd_def)
2500         .insert("status", status_cmd_def)
2501         .insert("key", key_mgmt_cli())
2502         .insert("mount", mount_cmd_def)
2503         .insert("catalog", catalog_mgmt_cli())
2504         .insert("task", task_mgmt_cli());
2505
2506     let rpcenv = CliEnvironment::new();
2507     run_cli_command(cmd_def, rpcenv, Some(|future| {
2508         proxmox_backup::tools::runtime::main(future)
2509     }));
2510 }