src/bin/proxmox-backup-client.rs

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