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