src/bin/proxmox-backup-client.rs

   1 use std::collections::{HashSet, HashMap};
   2 use std::convert::TryFrom;
   3 use std::io::{self, Read, Write, Seek, SeekFrom};
   4 use std::os::unix::io::{FromRawFd, RawFd};
   5 use std::path::{Path, PathBuf};
   6 use std::pin::Pin;
   7 use std::sync::{Arc, Mutex};
   8 use std::task::Context;
   9
  10 use anyhow::{bail, format_err, Error};
  11 use chrono::{Local, DateTime, Utc, TimeZone};
  12 use futures::future::FutureExt;
  13 use futures::stream::{StreamExt, TryStreamExt};
  14 use serde_json::{json, Value};
  15 use tokio::sync::mpsc;
  16 use xdg::BaseDirectories;
  17
  18 use pathpatterns::{MatchEntry, MatchType, PatternFlag};
  19 use proxmox::tools::fs::{file_get_contents, file_get_json, replace_file, CreateOptions, image_size};
  20 use proxmox::api::{ApiHandler, ApiMethod, RpcEnvironment};
  21 use proxmox::api::schema::*;
  22 use proxmox::api::cli::*;
  23 use proxmox::api::api;
  24 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
  25
  26 use proxmox_backup::tools;
  27 use proxmox_backup::api2::types::*;
  28 use proxmox_backup::api2::version;
  29 use proxmox_backup::client::*;
  30 use proxmox_backup::pxar::catalog::*;
  31 use proxmox_backup::backup::{
  32     archive_type,
  33     decrypt_key,
  34     verify_chunk_size,
  35     ArchiveType,
  36     AsyncReadChunk,
  37     BackupDir,
  38     BackupGroup,
  39     BackupManifest,
  40     BufferedDynamicReader,
  41     CATALOG_NAME,
  42     CatalogReader,
  43     CatalogWriter,
  44     ChunkStream,
  45     CryptConfig,
  46     CryptMode,
  47     DataBlob,
  48     DynamicIndexReader,
  49     FixedChunkStream,
  50     FixedIndexReader,
  51     IndexFile,
  52     MANIFEST_BLOB_NAME,
  53     Shell,
  54 };
  55
  56 mod proxmox_backup_client;
  57 use proxmox_backup_client::*;
  58
  59 const ENV_VAR_PBS_FINGERPRINT: &str = "PBS_FINGERPRINT";
  60 const ENV_VAR_PBS_PASSWORD: &str = "PBS_PASSWORD";
  61
  62
  63 pub const REPO_URL_SCHEMA: Schema = StringSchema::new("Repository URL.")
  64     .format(&BACKUP_REPO_URL)
  65     .max_length(256)
  66     .schema();
  67
  68 pub const KEYFILE_SCHEMA: Schema = StringSchema::new(
  69     "Path to encryption key. All data will be encrypted using this key.")
  70     .schema();
  71
  72 pub const KEYFD_SCHEMA: Schema = IntegerSchema::new(
  73     "Pass an encryption key via an already opened file descriptor.")
  74     .minimum(0)
  75     .schema();
  76
  77 const CHUNK_SIZE_SCHEMA: Schema = IntegerSchema::new(
  78     "Chunk size in KB. Must be a power of 2.")
  79     .minimum(64)
  80     .maximum(4096)
  81     .default(4096)
  82     .schema();
  83
  84 fn get_default_repository() -> Option<String> {
  85     std::env::var("PBS_REPOSITORY").ok()
  86 }
  87
  88 pub fn extract_repository_from_value(
  89     param: &Value,
  90 ) -> Result<BackupRepository, Error> {
  91
  92     let repo_url = param["repository"]
  93         .as_str()
  94         .map(String::from)
  95         .or_else(get_default_repository)
  96         .ok_or_else(|| format_err!("unable to get (default) repository"))?;
  97
  98     let repo: BackupRepository = repo_url.parse()?;
  99
 100     Ok(repo)
 101 }
 102
 103 fn extract_repository_from_map(
 104     param: &HashMap<String, String>,
 105 ) -> Option<BackupRepository> {
 106
 107     param.get("repository")
 108         .map(String::from)
 109         .or_else(get_default_repository)
 110         .and_then(|repo_url| repo_url.parse::<BackupRepository>().ok())
 111 }
 112
 113 fn record_repository(repo: &BackupRepository) {
 114
 115     let base = match BaseDirectories::with_prefix("proxmox-backup") {
 116         Ok(v) => v,
 117         _ => return,
 118     };
 119
 120     // usually $HOME/.cache/proxmox-backup/repo-list
 121     let path = match base.place_cache_file("repo-list") {
 122         Ok(v) => v,
 123         _ => return,
 124     };
 125
 126     let mut data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
 127
 128     let repo = repo.to_string();
 129
 130     data[&repo] = json!{ data[&repo].as_i64().unwrap_or(0) + 1 };
 131
 132     let mut map = serde_json::map::Map::new();
 133
 134     loop {
 135         let mut max_used = 0;
 136         let mut max_repo = None;
 137         for (repo, count) in data.as_object().unwrap() {
 138             if map.contains_key(repo) { continue; }
 139             if let Some(count) = count.as_i64() {
 140                 if count > max_used {
 141                     max_used = count;
 142                     max_repo = Some(repo);
 143                 }
 144             }
 145         }
 146         if let Some(repo) = max_repo {
 147             map.insert(repo.to_owned(), json!(max_used));
 148         } else {
 149             break;
 150         }
 151         if map.len() > 10 { // store max. 10 repos
 152             break;
 153         }
 154     }
 155
 156     let new_data = json!(map);
 157
 158     let _ = replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new());
 159 }
 160
 161 pub fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
 162
 163     let mut result = vec![];
 164
 165     let base = match BaseDirectories::with_prefix("proxmox-backup") {
 166         Ok(v) => v,
 167         _ => return result,
 168     };
 169
 170     // usually $HOME/.cache/proxmox-backup/repo-list
 171     let path = match base.place_cache_file("repo-list") {
 172         Ok(v) => v,
 173         _ => return result,
 174     };
 175
 176     let data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
 177
 178     if let Some(map) = data.as_object() {
 179         for (repo, _count) in map {
 180             result.push(repo.to_owned());
 181         }
 182     }
 183
 184     result
 185 }
 186
 187 fn connect(server: &str, userid: &str) -> Result<HttpClient, Error> {
 188
 189     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
 190
 191     use std::env::VarError::*;
 192     let password = match std::env::var(ENV_VAR_PBS_PASSWORD) {
 193         Ok(p) => Some(p),
 194         Err(NotUnicode(_)) => bail!(format!("{} contains bad characters", ENV_VAR_PBS_PASSWORD)),
 195         Err(NotPresent) => None,
 196     };
 197
 198     let options = HttpClientOptions::new()
 199         .prefix(Some("proxmox-backup".to_string()))
 200         .password(password)
 201         .interactive(true)
 202         .fingerprint(fingerprint)
 203         .fingerprint_cache(true)
 204         .ticket_cache(true);
 205
 206     HttpClient::new(server, userid, options)
 207 }
 208
 209 async fn view_task_result(
 210     client: HttpClient,
 211     result: Value,
 212     output_format: &str,
 213 ) -> Result<(), Error> {
 214     let data = &result["data"];
 215     if output_format == "text" {
 216         if let Some(upid) = data.as_str() {
 217             display_task_log(client, upid, true).await?;
 218         }
 219     } else {
 220         format_and_print_result(&data, &output_format);
 221     }
 222
 223     Ok(())
 224 }
 225
 226 async fn api_datastore_list_snapshots(
 227     client: &HttpClient,
 228     store: &str,
 229     group: Option<BackupGroup>,
 230 ) -> Result<Value, Error> {
 231
 232     let path = format!("api2/json/admin/datastore/{}/snapshots", store);
 233
 234     let mut args = json!({});
 235     if let Some(group) = group {
 236         args["backup-type"] = group.backup_type().into();
 237         args["backup-id"] = group.backup_id().into();
 238     }
 239
 240     let mut result = client.get(&path, Some(args)).await?;
 241
 242     Ok(result["data"].take())
 243 }
 244
 245 pub async fn api_datastore_latest_snapshot(
 246     client: &HttpClient,
 247     store: &str,
 248     group: BackupGroup,
 249 ) -> Result<(String, String, DateTime<Utc>), Error> {
 250
 251     let list = api_datastore_list_snapshots(client, store, Some(group.clone())).await?;
 252     let mut list: Vec<SnapshotListItem> = serde_json::from_value(list)?;
 253
 254     if list.is_empty() {
 255         bail!("backup group {:?} does not contain any snapshots.", group.group_path());
 256     }
 257
 258     list.sort_unstable_by(|a, b| b.backup_time.cmp(&a.backup_time));
 259
 260     let backup_time = Utc.timestamp(list[0].backup_time, 0);
 261
 262     Ok((group.backup_type().to_owned(), group.backup_id().to_owned(), backup_time))
 263 }
 264
 265 async fn backup_directory<P: AsRef<Path>>(
 266     client: &BackupWriter,
 267     previous_manifest: Option<Arc<BackupManifest>>,
 268     dir_path: P,
 269     archive_name: &str,
 270     chunk_size: Option<usize>,
 271     device_set: Option<HashSet<u64>>,
 272     verbose: bool,
 273     skip_lost_and_found: bool,
 274     catalog: Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 275     exclude_pattern: Vec<MatchEntry>,
 276     entries_max: usize,
 277     compress: bool,
 278     encrypt: bool,
 279 ) -> Result<BackupStats, Error> {
 280
 281     let pxar_stream = PxarBackupStream::open(
 282         dir_path.as_ref(),
 283         device_set,
 284         verbose,
 285         skip_lost_and_found,
 286         catalog,
 287         exclude_pattern,
 288         entries_max,
 289     )?;
 290     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 291
 292     let (mut tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 293
 294     let stream = rx
 295         .map_err(Error::from);
 296
 297     // spawn chunker inside a separate task so that it can run parallel
 298     tokio::spawn(async move {
 299         while let Some(v) = chunk_stream.next().await {
 300             let _ = tx.send(v).await;
 301         }
 302     });
 303
 304     let stats = client
 305         .upload_stream(previous_manifest, archive_name, stream, "dynamic", None, compress, encrypt)
 306         .await?;
 307
 308     Ok(stats)
 309 }
 310
 311 async fn backup_image<P: AsRef<Path>>(
 312     client: &BackupWriter,
 313     previous_manifest: Option<Arc<BackupManifest>>,
 314     image_path: P,
 315     archive_name: &str,
 316     image_size: u64,
 317     chunk_size: Option<usize>,
 318     compress: bool,
 319     encrypt: bool,
 320     _verbose: bool,
 321 ) -> Result<BackupStats, Error> {
 322
 323     let path = image_path.as_ref().to_owned();
 324
 325     let file = tokio::fs::File::open(path).await?;
 326
 327     let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 328         .map_err(Error::from);
 329
 330     let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
 331
 332     let stats = client
 333         .upload_stream(previous_manifest, archive_name, stream, "fixed", Some(image_size), compress, encrypt)
 334         .await?;
 335
 336     Ok(stats)
 337 }
 338
 339 #[api(
 340    input: {
 341         properties: {
 342             repository: {
 343                 schema: REPO_URL_SCHEMA,
 344                 optional: true,
 345             },
 346             "output-format": {
 347                 schema: OUTPUT_FORMAT,
 348                 optional: true,
 349             },
 350         }
 351    }
 352 )]
 353 /// List backup groups.
 354 async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 355
 356     let output_format = get_output_format(&param);
 357
 358     let repo = extract_repository_from_value(&param)?;
 359
 360     let client = connect(repo.host(), repo.user())?;
 361
 362     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 363
 364     let mut result = client.get(&path, None).await?;
 365
 366     record_repository(&repo);
 367
 368     let render_group_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 369         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 370         let group = BackupGroup::new(item.backup_type, item.backup_id);
 371         Ok(group.group_path().to_str().unwrap().to_owned())
 372     };
 373
 374     let render_last_backup = |_v: &Value, record: &Value| -> Result<String, Error> {
 375         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 376         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.last_backup);
 377         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 378     };
 379
 380     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 381         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 382         Ok(tools::format::render_backup_file_list(&item.files))
 383     };
 384
 385     let options = default_table_format_options()
 386         .sortby("backup-type", false)
 387         .sortby("backup-id", false)
 388         .column(ColumnConfig::new("backup-id").renderer(render_group_path).header("group"))
 389         .column(
 390             ColumnConfig::new("last-backup")
 391                 .renderer(render_last_backup)
 392                 .header("last snapshot")
 393                 .right_align(false)
 394         )
 395         .column(ColumnConfig::new("backup-count"))
 396         .column(ColumnConfig::new("files").renderer(render_files));
 397
 398     let mut data: Value = result["data"].take();
 399
 400     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_GROUPS;
 401
 402     format_and_print_result_full(&mut data, info, &output_format, &options);
 403
 404     Ok(Value::Null)
 405 }
 406
 407 #[api(
 408    input: {
 409         properties: {
 410             repository: {
 411                 schema: REPO_URL_SCHEMA,
 412                 optional: true,
 413             },
 414             group: {
 415                 type: String,
 416                 description: "Backup group.",
 417                 optional: true,
 418             },
 419             "output-format": {
 420                 schema: OUTPUT_FORMAT,
 421                 optional: true,
 422             },
 423         }
 424    }
 425 )]
 426 /// List backup snapshots.
 427 async fn list_snapshots(param: Value) -> Result<Value, Error> {
 428
 429     let repo = extract_repository_from_value(&param)?;
 430
 431     let output_format = get_output_format(&param);
 432
 433     let client = connect(repo.host(), repo.user())?;
 434
 435     let group: Option<BackupGroup> = if let Some(path) = param["group"].as_str() {
 436         Some(path.parse()?)
 437     } else {
 438         None
 439     };
 440
 441     let mut data = api_datastore_list_snapshots(&client, repo.store(), group).await?;
 442
 443     record_repository(&repo);
 444
 445     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 446         let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
 447         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time);
 448         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 449     };
 450
 451     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 452         let item: SnapshotListItem = serde_json::from_value(record.to_owned())?;
 453         let mut filenames = Vec::new();
 454         for file in &item.files {
 455             filenames.push(file.filename.to_string());
 456         }
 457         Ok(tools::format::render_backup_file_list(&filenames[..]))
 458     };
 459
 460     let options = default_table_format_options()
 461         .sortby("backup-type", false)
 462         .sortby("backup-id", false)
 463         .sortby("backup-time", false)
 464         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
 465         .column(ColumnConfig::new("size"))
 466         .column(ColumnConfig::new("files").renderer(render_files))
 467         ;
 468
 469     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOTS;
 470
 471     format_and_print_result_full(&mut data, info, &output_format, &options);
 472
 473     Ok(Value::Null)
 474 }
 475
 476 #[api(
 477    input: {
 478         properties: {
 479             repository: {
 480                 schema: REPO_URL_SCHEMA,
 481                 optional: true,
 482             },
 483             snapshot: {
 484                 type: String,
 485                 description: "Snapshot path.",
 486              },
 487         }
 488    }
 489 )]
 490 /// Forget (remove) backup snapshots.
 491 async fn forget_snapshots(param: Value) -> Result<Value, Error> {
 492
 493     let repo = extract_repository_from_value(&param)?;
 494
 495     let path = tools::required_string_param(&param, "snapshot")?;
 496     let snapshot: BackupDir = path.parse()?;
 497
 498     let mut client = connect(repo.host(), repo.user())?;
 499
 500     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
 501
 502     let result = client.delete(&path, Some(json!({
 503         "backup-type": snapshot.group().backup_type(),
 504         "backup-id": snapshot.group().backup_id(),
 505         "backup-time": snapshot.backup_time().timestamp(),
 506     }))).await?;
 507
 508     record_repository(&repo);
 509
 510     Ok(result)
 511 }
 512
 513 #[api(
 514    input: {
 515         properties: {
 516             repository: {
 517                 schema: REPO_URL_SCHEMA,
 518                 optional: true,
 519             },
 520         }
 521    }
 522 )]
 523 /// Try to login. If successful, store ticket.
 524 async fn api_login(param: Value) -> Result<Value, Error> {
 525
 526     let repo = extract_repository_from_value(&param)?;
 527
 528     let client = connect(repo.host(), repo.user())?;
 529     client.login().await?;
 530
 531     record_repository(&repo);
 532
 533     Ok(Value::Null)
 534 }
 535
 536 #[api(
 537    input: {
 538         properties: {
 539             repository: {
 540                 schema: REPO_URL_SCHEMA,
 541                 optional: true,
 542             },
 543         }
 544    }
 545 )]
 546 /// Logout (delete stored ticket).
 547 fn api_logout(param: Value) -> Result<Value, Error> {
 548
 549     let repo = extract_repository_from_value(&param)?;
 550
 551     delete_ticket_info("proxmox-backup", repo.host(), repo.user())?;
 552
 553     Ok(Value::Null)
 554 }
 555
 556 #[api(
 557    input: {
 558         properties: {
 559             repository: {
 560                 schema: REPO_URL_SCHEMA,
 561                 optional: true,
 562             },
 563             "output-format": {
 564                 schema: OUTPUT_FORMAT,
 565                 optional: true,
 566             },
 567         }
 568    }
 569 )]
 570 /// Show client and optional server version
 571 async fn api_version(param: Value) -> Result<(), Error> {
 572
 573     let output_format = get_output_format(&param);
 574
 575     let mut version_info = json!({
 576         "client": {
 577             "version": version::PROXMOX_PKG_VERSION,
 578             "release": version::PROXMOX_PKG_RELEASE,
 579             "repoid": version::PROXMOX_PKG_REPOID,
 580         }
 581     });
 582
 583     let repo = extract_repository_from_value(&param);
 584     if let Ok(repo) = repo {
 585         let client = connect(repo.host(), repo.user())?;
 586
 587         match client.get("api2/json/version", None).await {
 588             Ok(mut result) => version_info["server"] = result["data"].take(),
 589             Err(e) => eprintln!("could not connect to server - {}", e),
 590         }
 591     }
 592     if output_format == "text" {
 593         println!("client version: {}.{}", version::PROXMOX_PKG_VERSION, version::PROXMOX_PKG_RELEASE);
 594         if let Some(server) = version_info["server"].as_object() {
 595             let server_version = server["version"].as_str().unwrap();
 596             let server_release = server["release"].as_str().unwrap();
 597             println!("server version: {}.{}", server_version, server_release);
 598         }
 599     } else {
 600         format_and_print_result(&version_info, &output_format);
 601     }
 602
 603     Ok(())
 604 }
 605
 606
 607 #[api(
 608    input: {
 609         properties: {
 610             repository: {
 611                 schema: REPO_URL_SCHEMA,
 612                 optional: true,
 613             },
 614             snapshot: {
 615                 type: String,
 616                 description: "Snapshot path.",
 617              },
 618             "output-format": {
 619                 schema: OUTPUT_FORMAT,
 620                 optional: true,
 621             },
 622         }
 623    }
 624 )]
 625 /// List snapshot files.
 626 async fn list_snapshot_files(param: Value) -> Result<Value, Error> {
 627
 628     let repo = extract_repository_from_value(&param)?;
 629
 630     let path = tools::required_string_param(&param, "snapshot")?;
 631     let snapshot: BackupDir = path.parse()?;
 632
 633     let output_format = get_output_format(&param);
 634
 635     let client = connect(repo.host(), repo.user())?;
 636
 637     let path = format!("api2/json/admin/datastore/{}/files", repo.store());
 638
 639     let mut result = client.get(&path, Some(json!({
 640         "backup-type": snapshot.group().backup_type(),
 641         "backup-id": snapshot.group().backup_id(),
 642         "backup-time": snapshot.backup_time().timestamp(),
 643     }))).await?;
 644
 645     record_repository(&repo);
 646
 647     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_LIST_SNAPSHOT_FILES;
 648
 649     let mut data: Value = result["data"].take();
 650
 651     let options = default_table_format_options();
 652
 653     format_and_print_result_full(&mut data, info, &output_format, &options);
 654
 655     Ok(Value::Null)
 656 }
 657
 658 #[api(
 659     input: {
 660         properties: {
 661             repository: {
 662                 schema: REPO_URL_SCHEMA,
 663                 optional: true,
 664             },
 665             "output-format": {
 666                 schema: OUTPUT_FORMAT,
 667                 optional: true,
 668             },
 669         },
 670     },
 671 )]
 672 /// Start garbage collection for a specific repository.
 673 async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 674
 675     let repo = extract_repository_from_value(&param)?;
 676
 677     let output_format = get_output_format(&param);
 678
 679     let mut client = connect(repo.host(), repo.user())?;
 680
 681     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 682
 683     let result = client.post(&path, None).await?;
 684
 685     record_repository(&repo);
 686
 687     view_task_result(client, result, &output_format).await?;
 688
 689     Ok(Value::Null)
 690 }
 691
 692 fn spawn_catalog_upload(
 693     client: Arc<BackupWriter>,
 694     encrypt: bool,
 695 ) -> Result<
 696         (
 697             Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 698             tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>
 699         ), Error>
 700 {
 701     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
 702     let catalog_stream = crate::tools::StdChannelStream(catalog_rx);
 703     let catalog_chunk_size = 512*1024;
 704     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 705
 706     let catalog = Arc::new(Mutex::new(CatalogWriter::new(crate::tools::StdChannelWriter::new(catalog_tx))?));
 707
 708     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 709
 710     tokio::spawn(async move {
 711         let catalog_upload_result = client
 712             .upload_stream(None, CATALOG_NAME, catalog_chunk_stream, "dynamic", None, true, encrypt)
 713             .await;
 714
 715         if let Err(ref err) = catalog_upload_result {
 716             eprintln!("catalog upload error - {}", err);
 717             client.cancel();
 718         }
 719
 720         let _ = catalog_result_tx.send(catalog_upload_result);
 721     });
 722
 723     Ok((catalog, catalog_result_rx))
 724 }
 725
 726 fn keyfile_parameters(param: &Value) -> Result<(Option<Vec<u8>>, CryptMode), Error> {
 727     let keyfile = match param.get("keyfile") {
 728         Some(Value::String(keyfile)) => Some(keyfile),
 729         Some(_) => bail!("bad --keyfile parameter type"),
 730         None => None,
 731     };
 732
 733     let key_fd = match param.get("keyfd") {
 734         Some(Value::Number(key_fd)) => Some(
 735             RawFd::try_from(key_fd
 736                 .as_i64()
 737                 .ok_or_else(|| format_err!("bad key fd: {:?}", key_fd))?
 738             )
 739             .map_err(|err| format_err!("bad key fd: {:?}: {}", key_fd, err))?
 740         ),
 741         Some(_) => bail!("bad --keyfd parameter type"),
 742         None => None,
 743     };
 744
 745     let crypt_mode: Option<CryptMode> = match param.get("crypt-mode") {
 746         Some(mode) => Some(serde_json::from_value(mode.clone())?),
 747         None => None,
 748     };
 749
 750     let keydata = match (keyfile, key_fd) {
 751         (None, None) => None,
 752         (Some(_), Some(_)) => bail!("--keyfile and --keyfd are mutually exclusive"),
 753         (Some(keyfile), None) => Some(file_get_contents(keyfile)?),
 754         (None, Some(fd)) => {
 755             let input = unsafe { std::fs::File::from_raw_fd(fd) };
 756             let mut data = Vec::new();
 757             let _len: usize = { input }.read_to_end(&mut data)
 758                 .map_err(|err| {
 759                     format_err!("error reading encryption key from fd {}: {}", fd, err)
 760                 })?;
 761             Some(data)
 762         }
 763     };
 764
 765     Ok(match (keydata, crypt_mode) {
 766         // no parameters:
 767         (None, None) => match key::read_optional_default_encryption_key()? {
 768             Some(key) => (Some(key), CryptMode::Encrypt),
 769             None => (None, CryptMode::None),
 770         },
 771
 772         // just --crypt-mode=none
 773         (None, Some(CryptMode::None)) => (None, CryptMode::None),
 774
 775         // just --crypt-mode other than none
 776         (None, Some(crypt_mode)) => match key::read_optional_default_encryption_key()? {
 777             None => bail!("--crypt-mode without --keyfile and no default key file available"),
 778             Some(key) => (Some(key), crypt_mode),
 779         }
 780
 781         // just --keyfile
 782         (Some(key), None) => (Some(key), CryptMode::Encrypt),
 783
 784         // --keyfile and --crypt-mode=none
 785         (Some(_), Some(CryptMode::None)) => {
 786             bail!("--keyfile/--keyfd and --crypt-mode=none are mutually exclusive");
 787         }
 788
 789         // --keyfile and --crypt-mode other than none
 790         (Some(key), Some(crypt_mode)) => (Some(key), crypt_mode),
 791     })
 792 }
 793
 794 #[api(
 795    input: {
 796        properties: {
 797            backupspec: {
 798                type: Array,
 799                description: "List of backup source specifications ([<label.ext>:<path>] ...)",
 800                items: {
 801                    schema: BACKUP_SOURCE_SCHEMA,
 802                }
 803            },
 804            repository: {
 805                schema: REPO_URL_SCHEMA,
 806                optional: true,
 807            },
 808            "include-dev": {
 809                description: "Include mountpoints with same st_dev number (see ``man fstat``) as specified files.",
 810                optional: true,
 811                items: {
 812                    type: String,
 813                    description: "Path to file.",
 814                }
 815            },
 816            keyfile: {
 817                schema: KEYFILE_SCHEMA,
 818                optional: true,
 819            },
 820            "keyfd": {
 821                schema: KEYFD_SCHEMA,
 822                optional: true,
 823            },
 824            "crypt-mode": {
 825                type: CryptMode,
 826                optional: true,
 827            },
 828            "skip-lost-and-found": {
 829                type: Boolean,
 830                description: "Skip lost+found directory.",
 831                optional: true,
 832            },
 833            "backup-type": {
 834                schema: BACKUP_TYPE_SCHEMA,
 835                optional: true,
 836            },
 837            "backup-id": {
 838                schema: BACKUP_ID_SCHEMA,
 839                optional: true,
 840            },
 841            "backup-time": {
 842                schema: BACKUP_TIME_SCHEMA,
 843                optional: true,
 844            },
 845            "chunk-size": {
 846                schema: CHUNK_SIZE_SCHEMA,
 847                optional: true,
 848            },
 849            "exclude": {
 850                type: Array,
 851                description: "List of paths or patterns for matching files to exclude.",
 852                optional: true,
 853                items: {
 854                    type: String,
 855                    description: "Path or match pattern.",
 856                 }
 857            },
 858            "entries-max": {
 859                type: Integer,
 860                description: "Max number of entries to hold in memory.",
 861                optional: true,
 862                default: proxmox_backup::pxar::ENCODER_MAX_ENTRIES as isize,
 863            },
 864            "verbose": {
 865                type: Boolean,
 866                description: "Verbose output.",
 867                optional: true,
 868            },
 869        }
 870    }
 871 )]
 872 /// Create (host) backup.
 873 async fn create_backup(
 874     param: Value,
 875     _info: &ApiMethod,
 876     _rpcenv: &mut dyn RpcEnvironment,
 877 ) -> Result<Value, Error> {
 878
 879     let repo = extract_repository_from_value(&param)?;
 880
 881     let backupspec_list = tools::required_array_param(&param, "backupspec")?;
 882
 883     let all_file_systems = param["all-file-systems"].as_bool().unwrap_or(false);
 884
 885     let skip_lost_and_found = param["skip-lost-and-found"].as_bool().unwrap_or(false);
 886
 887     let verbose = param["verbose"].as_bool().unwrap_or(false);
 888
 889     let backup_time_opt = param["backup-time"].as_i64();
 890
 891     let chunk_size_opt = param["chunk-size"].as_u64().map(|v| (v*1024) as usize);
 892
 893     if let Some(size) = chunk_size_opt {
 894         verify_chunk_size(size)?;
 895     }
 896
 897     let (keydata, crypt_mode) = keyfile_parameters(&param)?;
 898
 899     let backup_id = param["backup-id"].as_str().unwrap_or(&proxmox::tools::nodename());
 900
 901     let backup_type = param["backup-type"].as_str().unwrap_or("host");
 902
 903     let include_dev = param["include-dev"].as_array();
 904
 905     let entries_max = param["entries-max"].as_u64()
 906         .unwrap_or(proxmox_backup::pxar::ENCODER_MAX_ENTRIES as u64);
 907
 908     let empty = Vec::new();
 909     let exclude_args = param["exclude"].as_array().unwrap_or(&empty);
 910
 911     let mut pattern_list = Vec::with_capacity(exclude_args.len());
 912     for entry in exclude_args {
 913         let entry = entry.as_str().ok_or_else(|| format_err!("Invalid pattern string slice"))?;
 914         pattern_list.push(
 915             MatchEntry::parse_pattern(entry, PatternFlag::PATH_NAME, MatchType::Exclude)
 916                 .map_err(|err| format_err!("invalid exclude pattern entry: {}", err))?
 917         );
 918     }
 919
 920     let mut devices = if all_file_systems { None } else { Some(HashSet::new()) };
 921
 922     if let Some(include_dev) = include_dev {
 923         if all_file_systems {
 924             bail!("option 'all-file-systems' conflicts with option 'include-dev'");
 925         }
 926
 927         let mut set = HashSet::new();
 928         for path in include_dev {
 929             let path = path.as_str().unwrap();
 930             let stat = nix::sys::stat::stat(path)
 931                 .map_err(|err| format_err!("fstat {:?} failed - {}", path, err))?;
 932             set.insert(stat.st_dev);
 933         }
 934         devices = Some(set);
 935     }
 936
 937     let mut upload_list = vec![];
 938
 939     for backupspec in backupspec_list {
 940         let spec = parse_backup_specification(backupspec.as_str().unwrap())?;
 941         let filename = &spec.config_string;
 942         let target = &spec.archive_name;
 943
 944         use std::os::unix::fs::FileTypeExt;
 945
 946         let metadata = std::fs::metadata(filename)
 947             .map_err(|err| format_err!("unable to access '{}' - {}", filename, err))?;
 948         let file_type = metadata.file_type();
 949
 950         match spec.spec_type {
 951             BackupSpecificationType::PXAR => {
 952                 if !file_type.is_dir() {
 953                     bail!("got unexpected file type (expected directory)");
 954                 }
 955                 upload_list.push((BackupSpecificationType::PXAR, filename.to_owned(), format!("{}.didx", target), 0));
 956             }
 957             BackupSpecificationType::IMAGE => {
 958                 if !(file_type.is_file() || file_type.is_block_device()) {
 959                     bail!("got unexpected file type (expected file or block device)");
 960                 }
 961
 962                 let size = image_size(&PathBuf::from(filename))?;
 963
 964                 if size == 0 { bail!("got zero-sized file '{}'", filename); }
 965
 966                 upload_list.push((BackupSpecificationType::IMAGE, filename.to_owned(), format!("{}.fidx", target), size));
 967             }
 968             BackupSpecificationType::CONFIG => {
 969                 if !file_type.is_file() {
 970                     bail!("got unexpected file type (expected regular file)");
 971                 }
 972                 upload_list.push((BackupSpecificationType::CONFIG, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 973             }
 974             BackupSpecificationType::LOGFILE => {
 975                 if !file_type.is_file() {
 976                     bail!("got unexpected file type (expected regular file)");
 977                 }
 978                 upload_list.push((BackupSpecificationType::LOGFILE, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 979             }
 980         }
 981     }
 982
 983     let backup_time = Utc.timestamp(backup_time_opt.unwrap_or_else(|| Utc::now().timestamp()), 0);
 984
 985     let client = connect(repo.host(), repo.user())?;
 986     record_repository(&repo);
 987
 988     println!("Starting backup: {}/{}/{}", backup_type, backup_id, BackupDir::backup_time_to_string(backup_time));
 989
 990     println!("Client name: {}", proxmox::tools::nodename());
 991
 992     let start_time = Local::now();
 993
 994     println!("Starting protocol: {}", start_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, false));
 995
 996     let (crypt_config, rsa_encrypted_key) = match keydata {
 997         None => (None, None),
 998         Some(key) => {
 999             let (key, created) = decrypt_key(&key, &key::get_encryption_key_password)?;
1000
1001             let crypt_config = CryptConfig::new(key)?;
1002
1003             match key::find_master_pubkey()? {
1004                 Some(ref path) if path.exists() => {
1005                     let pem_data = file_get_contents(path)?;
1006                     let rsa = openssl::rsa::Rsa::public_key_from_pem(&pem_data)?;
1007                     let enc_key = crypt_config.generate_rsa_encoded_key(rsa, created)?;
1008                     (Some(Arc::new(crypt_config)), Some(enc_key))
1009                 }
1010                 _ => (Some(Arc::new(crypt_config)), None),
1011             }
1012         }
1013     };
1014
1015     let client = BackupWriter::start(
1016         client,
1017         crypt_config.clone(),
1018         repo.store(),
1019         backup_type,
1020         &backup_id,
1021         backup_time,
1022         verbose,
1023     ).await?;
1024
1025     let previous_manifest = if let Ok(previous_manifest) = client.download_previous_manifest().await {
1026         Some(Arc::new(previous_manifest))
1027     } else {
1028         None
1029     };
1030
1031     let snapshot = BackupDir::new(backup_type, backup_id, backup_time.timestamp());
1032     let mut manifest = BackupManifest::new(snapshot);
1033
1034     let mut catalog = None;
1035     let mut catalog_result_tx = None;
1036
1037     for (backup_type, filename, target, size) in upload_list {
1038         match backup_type {
1039             BackupSpecificationType::CONFIG => {
1040                 println!("Upload config file '{}' to '{}' as {}", filename, repo, target);
1041                 let stats = client
1042                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1043                     .await?;
1044                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1045             }
1046             BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
1047                 println!("Upload log file '{}' to '{}' as {}", filename, repo, target);
1048                 let stats = client
1049                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1050                     .await?;
1051                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1052             }
1053             BackupSpecificationType::PXAR => {
1054                 // start catalog upload on first use
1055                 if catalog.is_none() {
1056                     let (cat, res) = spawn_catalog_upload(client.clone(), crypt_mode == CryptMode::Encrypt)?;
1057                     catalog = Some(cat);
1058                     catalog_result_tx = Some(res);
1059                 }
1060                 let catalog = catalog.as_ref().unwrap();
1061
1062                 println!("Upload directory '{}' to '{}' as {}", filename, repo, target);
1063                 catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
1064                 let stats = backup_directory(
1065                     &client,
1066                     previous_manifest.clone(),
1067                     &filename,
1068                     &target,
1069                     chunk_size_opt,
1070                     devices.clone(),
1071                     verbose,
1072                     skip_lost_and_found,
1073                     catalog.clone(),
1074                     pattern_list.clone(),
1075                     entries_max as usize,
1076                     true,
1077                     crypt_mode == CryptMode::Encrypt,
1078                 ).await?;
1079                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1080                 catalog.lock().unwrap().end_directory()?;
1081             }
1082             BackupSpecificationType::IMAGE => {
1083                 println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
1084                 let stats = backup_image(
1085                     &client,
1086                     previous_manifest.clone(),
1087                      &filename,
1088                     &target,
1089                     size,
1090                     chunk_size_opt,
1091                     true,
1092                     crypt_mode == CryptMode::Encrypt,
1093                     verbose,
1094                 ).await?;
1095                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1096             }
1097         }
1098     }
1099
1100     // finalize and upload catalog
1101     if let Some(catalog) = catalog {
1102         let mutex = Arc::try_unwrap(catalog)
1103             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
1104         let mut catalog = mutex.into_inner().unwrap();
1105
1106         catalog.finish()?;
1107
1108         drop(catalog); // close upload stream
1109
1110         if let Some(catalog_result_rx) = catalog_result_tx {
1111             let stats = catalog_result_rx.await??;
1112             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypt_mode)?;
1113         }
1114     }
1115
1116     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
1117         let target = "rsa-encrypted.key";
1118         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
1119         let stats = client
1120             .upload_blob_from_data(rsa_encrypted_key, target, false, false)
1121             .await?;
1122         manifest.add_file(format!("{}.blob", target), stats.size, stats.csum, crypt_mode)?;
1123
1124         // openssl rsautl -decrypt -inkey master-private.pem -in rsa-encrypted.key -out t
1125         /*
1126         let mut buffer2 = vec![0u8; rsa.size() as usize];
1127         let pem_data = file_get_contents("master-private.pem")?;
1128         let rsa = openssl::rsa::Rsa::private_key_from_pem(&pem_data)?;
1129         let len = rsa.private_decrypt(&buffer, &mut buffer2, openssl::rsa::Padding::PKCS1)?;
1130         println!("TEST {} {:?}", len, buffer2);
1131          */
1132     }
1133
1134     // create manifest (index.json)
1135     // manifests are never encrypted, but include a signature
1136     let manifest = manifest.to_string(crypt_config.as_ref().map(Arc::as_ref))
1137         .map_err(|err| format_err!("unable to format manifest - {}", err))?;
1138
1139
1140     println!("Upload index.json to '{}'", repo);
1141     client
1142         .upload_blob_from_data(manifest.into_bytes(), MANIFEST_BLOB_NAME, true, false)
1143         .await?;
1144
1145     client.finish().await?;
1146
1147     let end_time = Local::now();
1148     let elapsed = end_time.signed_duration_since(start_time);
1149     println!("Duration: {}", elapsed);
1150
1151     println!("End Time: {}", end_time.to_rfc3339_opts(chrono::SecondsFormat::Secs, false));
1152
1153     Ok(Value::Null)
1154 }
1155
1156 fn complete_backup_source(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1157
1158     let mut result = vec![];
1159
1160     let data: Vec<&str> = arg.splitn(2, ':').collect();
1161
1162     if data.len() != 2 {
1163         result.push(String::from("root.pxar:/"));
1164         result.push(String::from("etc.pxar:/etc"));
1165         return result;
1166     }
1167
1168     let files = tools::complete_file_name(data[1], param);
1169
1170     for file in files {
1171         result.push(format!("{}:{}", data[0], file));
1172     }
1173
1174     result
1175 }
1176
1177 async fn dump_image<W: Write>(
1178     client: Arc<BackupReader>,
1179     crypt_config: Option<Arc<CryptConfig>>,
1180     index: FixedIndexReader,
1181     mut writer: W,
1182     verbose: bool,
1183 ) -> Result<(), Error> {
1184
1185     let most_used = index.find_most_used_chunks(8);
1186
1187     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
1188
1189     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
1190     // and thus slows down reading. Instead, directly use RemoteChunkReader
1191     let mut per = 0;
1192     let mut bytes = 0;
1193     let start_time = std::time::Instant::now();
1194
1195     for pos in 0..index.index_count() {
1196         let digest = index.index_digest(pos).unwrap();
1197         let raw_data = chunk_reader.read_chunk(&digest).await?;
1198         writer.write_all(&raw_data)?;
1199         bytes += raw_data.len();
1200         if verbose {
1201             let next_per = ((pos+1)*100)/index.index_count();
1202             if per != next_per {
1203                 eprintln!("progress {}% (read {} bytes, duration {} sec)",
1204                           next_per, bytes, start_time.elapsed().as_secs());
1205                 per = next_per;
1206             }
1207         }
1208     }
1209
1210     let end_time = std::time::Instant::now();
1211     let elapsed = end_time.duration_since(start_time);
1212     eprintln!("restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1213               bytes,
1214               elapsed.as_secs_f64(),
1215               bytes as f64/(1024.0*1024.0*elapsed.as_secs_f64())
1216     );
1217
1218
1219     Ok(())
1220 }
1221
1222 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1223     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1224         (name.into(), archive_type(name).unwrap())
1225     } else if name.ends_with(".pxar") {
1226         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1227     } else if name.ends_with(".img") {
1228         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1229     } else {
1230         (format!("{}.blob", name), ArchiveType::Blob)
1231     }
1232 }
1233
1234 #[api(
1235    input: {
1236        properties: {
1237            repository: {
1238                schema: REPO_URL_SCHEMA,
1239                optional: true,
1240            },
1241            snapshot: {
1242                type: String,
1243                description: "Group/Snapshot path.",
1244            },
1245            "archive-name": {
1246                description: "Backup archive name.",
1247                type: String,
1248            },
1249            target: {
1250                type: String,
1251                description: r###"Target directory path. Use '-' to write to standard output.
1252
1253 We do not extraxt '.pxar' archives when writing to standard output.
1254
1255 "###
1256            },
1257            "allow-existing-dirs": {
1258                type: Boolean,
1259                description: "Do not fail if directories already exists.",
1260                optional: true,
1261            },
1262            keyfile: {
1263                schema: KEYFILE_SCHEMA,
1264                optional: true,
1265            },
1266            "keyfd": {
1267                schema: KEYFD_SCHEMA,
1268                optional: true,
1269            },
1270            "crypt-mode": {
1271                type: CryptMode,
1272                optional: true,
1273            },
1274        }
1275    }
1276 )]
1277 /// Restore backup repository.
1278 async fn restore(param: Value) -> Result<Value, Error> {
1279     let repo = extract_repository_from_value(&param)?;
1280
1281     let verbose = param["verbose"].as_bool().unwrap_or(false);
1282
1283     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1284
1285     let archive_name = tools::required_string_param(&param, "archive-name")?;
1286
1287     let client = connect(repo.host(), repo.user())?;
1288
1289     record_repository(&repo);
1290
1291     let path = tools::required_string_param(&param, "snapshot")?;
1292
1293     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
1294         let group: BackupGroup = path.parse()?;
1295         api_datastore_latest_snapshot(&client, repo.store(), group).await?
1296     } else {
1297         let snapshot: BackupDir = path.parse()?;
1298         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
1299     };
1300
1301     let target = tools::required_string_param(&param, "target")?;
1302     let target = if target == "-" { None } else { Some(target) };
1303
1304     let (keydata, _crypt_mode) = keyfile_parameters(&param)?;
1305
1306     let crypt_config = match keydata {
1307         None => None,
1308         Some(key) => {
1309             let (key, _) = decrypt_key(&key, &key::get_encryption_key_password)?;
1310             Some(Arc::new(CryptConfig::new(key)?))
1311         }
1312     };
1313
1314     let client = BackupReader::start(
1315         client,
1316         crypt_config.clone(),
1317         repo.store(),
1318         &backup_type,
1319         &backup_id,
1320         backup_time,
1321         true,
1322     ).await?;
1323
1324     let (manifest, backup_index_data) = client.download_manifest().await?;
1325
1326     let (archive_name, archive_type) = parse_archive_type(archive_name);
1327
1328     if archive_name == MANIFEST_BLOB_NAME {
1329         if let Some(target) = target {
1330             replace_file(target, &backup_index_data, CreateOptions::new())?;
1331         } else {
1332             let stdout = std::io::stdout();
1333             let mut writer = stdout.lock();
1334             writer.write_all(&backup_index_data)
1335                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1336         }
1337
1338     } else if archive_type == ArchiveType::Blob {
1339
1340         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1341
1342         if let Some(target) = target {
1343            let mut writer = std::fs::OpenOptions::new()
1344                 .write(true)
1345                 .create(true)
1346                 .create_new(true)
1347                 .open(target)
1348                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?;
1349             std::io::copy(&mut reader, &mut writer)?;
1350         } else {
1351             let stdout = std::io::stdout();
1352             let mut writer = stdout.lock();
1353             std::io::copy(&mut reader, &mut writer)
1354                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1355         }
1356
1357     } else if archive_type == ArchiveType::DynamicIndex {
1358
1359         let index = client.download_dynamic_index(&manifest, &archive_name).await?;
1360
1361         let most_used = index.find_most_used_chunks(8);
1362
1363         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
1364
1365         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1366
1367         if let Some(target) = target {
1368             proxmox_backup::pxar::extract_archive(
1369                 pxar::decoder::Decoder::from_std(reader)?,
1370                 Path::new(target),
1371                 &[],
1372                 proxmox_backup::pxar::Flags::DEFAULT,
1373                 allow_existing_dirs,
1374                 |path| {
1375                     if verbose {
1376                         println!("{:?}", path);
1377                     }
1378                 },
1379             )
1380             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1381         } else {
1382             let mut writer = std::fs::OpenOptions::new()
1383                 .write(true)
1384                 .open("/dev/stdout")
1385                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1386
1387             std::io::copy(&mut reader, &mut writer)
1388                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1389         }
1390     } else if archive_type == ArchiveType::FixedIndex {
1391
1392         let index = client.download_fixed_index(&manifest, &archive_name).await?;
1393
1394         let mut writer = if let Some(target) = target {
1395             std::fs::OpenOptions::new()
1396                 .write(true)
1397                 .create(true)
1398                 .create_new(true)
1399                 .open(target)
1400                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1401         } else {
1402             std::fs::OpenOptions::new()
1403                 .write(true)
1404                 .open("/dev/stdout")
1405                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1406         };
1407
1408         dump_image(client.clone(), crypt_config.clone(), index, &mut writer, verbose).await?;
1409     }
1410
1411     Ok(Value::Null)
1412 }
1413
1414 #[api(
1415    input: {
1416        properties: {
1417            repository: {
1418                schema: REPO_URL_SCHEMA,
1419                optional: true,
1420            },
1421            snapshot: {
1422                type: String,
1423                description: "Group/Snapshot path.",
1424            },
1425            logfile: {
1426                type: String,
1427                description: "The path to the log file you want to upload.",
1428            },
1429            keyfile: {
1430                schema: KEYFILE_SCHEMA,
1431                optional: true,
1432            },
1433            "keyfd": {
1434                schema: KEYFD_SCHEMA,
1435                optional: true,
1436            },
1437            "crypt-mode": {
1438                type: CryptMode,
1439                optional: true,
1440            },
1441        }
1442    }
1443 )]
1444 /// Upload backup log file.
1445 async fn upload_log(param: Value) -> Result<Value, Error> {
1446
1447     let logfile = tools::required_string_param(&param, "logfile")?;
1448     let repo = extract_repository_from_value(&param)?;
1449
1450     let snapshot = tools::required_string_param(&param, "snapshot")?;
1451     let snapshot: BackupDir = snapshot.parse()?;
1452
1453     let mut client = connect(repo.host(), repo.user())?;
1454
1455     let (keydata, crypt_mode) = keyfile_parameters(&param)?;
1456
1457     let crypt_config = match keydata {
1458         None => None,
1459         Some(key) => {
1460             let (key, _created) = decrypt_key(&key, &key::get_encryption_key_password)?;
1461             let crypt_config = CryptConfig::new(key)?;
1462             Some(Arc::new(crypt_config))
1463         }
1464     };
1465
1466     let data = file_get_contents(logfile)?;
1467
1468     // fixme: howto sign log?
1469     let blob = match crypt_mode {
1470         CryptMode::None | CryptMode::SignOnly => DataBlob::encode(&data, None, true)?,
1471         CryptMode::Encrypt => DataBlob::encode(&data, crypt_config.as_ref().map(Arc::as_ref), true)?,
1472     };
1473
1474     let raw_data = blob.into_inner();
1475
1476     let path = format!("api2/json/admin/datastore/{}/upload-backup-log", repo.store());
1477
1478     let args = json!({
1479         "backup-type": snapshot.group().backup_type(),
1480         "backup-id":  snapshot.group().backup_id(),
1481         "backup-time": snapshot.backup_time().timestamp(),
1482     });
1483
1484     let body = hyper::Body::from(raw_data);
1485
1486     client.upload("application/octet-stream", body, &path, Some(args)).await
1487 }
1488
1489 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
1490     &ApiHandler::Async(&prune),
1491     &ObjectSchema::new(
1492         "Prune a backup repository.",
1493         &proxmox_backup::add_common_prune_prameters!([
1494             ("dry-run", true, &BooleanSchema::new(
1495                 "Just show what prune would do, but do not delete anything.")
1496              .schema()),
1497             ("group", false, &StringSchema::new("Backup group.").schema()),
1498         ], [
1499             ("output-format", true, &OUTPUT_FORMAT),
1500             (
1501                 "quiet",
1502                 true,
1503                 &BooleanSchema::new("Minimal output - only show removals.")
1504                     .schema()
1505             ),
1506             ("repository", true, &REPO_URL_SCHEMA),
1507         ])
1508     )
1509 );
1510
1511 fn prune<'a>(
1512     param: Value,
1513     _info: &ApiMethod,
1514     _rpcenv: &'a mut dyn RpcEnvironment,
1515 ) -> proxmox::api::ApiFuture<'a> {
1516     async move {
1517         prune_async(param).await
1518     }.boxed()
1519 }
1520
1521 async fn prune_async(mut param: Value) -> Result<Value, Error> {
1522     let repo = extract_repository_from_value(&param)?;
1523
1524     let mut client = connect(repo.host(), repo.user())?;
1525
1526     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1527
1528     let group = tools::required_string_param(&param, "group")?;
1529     let group: BackupGroup = group.parse()?;
1530
1531     let output_format = get_output_format(&param);
1532
1533     let quiet = param["quiet"].as_bool().unwrap_or(false);
1534
1535     param.as_object_mut().unwrap().remove("repository");
1536     param.as_object_mut().unwrap().remove("group");
1537     param.as_object_mut().unwrap().remove("output-format");
1538     param.as_object_mut().unwrap().remove("quiet");
1539
1540     param["backup-type"] = group.backup_type().into();
1541     param["backup-id"] = group.backup_id().into();
1542
1543     let mut result = client.post(&path, Some(param)).await?;
1544
1545     record_repository(&repo);
1546
1547     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1548         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1549         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time);
1550         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
1551     };
1552
1553     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1554         Ok(match v.as_bool() {
1555             Some(true) => "keep",
1556             Some(false) => "remove",
1557             None => "unknown",
1558         }.to_string())
1559     };
1560
1561     let options = default_table_format_options()
1562         .sortby("backup-type", false)
1563         .sortby("backup-id", false)
1564         .sortby("backup-time", false)
1565         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
1566         .column(ColumnConfig::new("backup-time").renderer(tools::format::render_epoch).header("date"))
1567         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
1568         ;
1569
1570     let info = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_PRUNE;
1571
1572     let mut data = result["data"].take();
1573
1574     if quiet {
1575         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
1576             item["keep"].as_bool() == Some(false)
1577         }).map(|v| v.clone()).collect();
1578         data = list.into();
1579     }
1580
1581     format_and_print_result_full(&mut data, info, &output_format, &options);
1582
1583     Ok(Value::Null)
1584 }
1585
1586 #[api(
1587    input: {
1588        properties: {
1589            repository: {
1590                schema: REPO_URL_SCHEMA,
1591                optional: true,
1592            },
1593            "output-format": {
1594                schema: OUTPUT_FORMAT,
1595                optional: true,
1596            },
1597        }
1598    }
1599 )]
1600 /// Get repository status.
1601 async fn status(param: Value) -> Result<Value, Error> {
1602
1603     let repo = extract_repository_from_value(&param)?;
1604
1605     let output_format = get_output_format(&param);
1606
1607     let client = connect(repo.host(), repo.user())?;
1608
1609     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1610
1611     let mut result = client.get(&path, None).await?;
1612     let mut data = result["data"].take();
1613
1614     record_repository(&repo);
1615
1616     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1617         let v = v.as_u64().unwrap();
1618         let total = record["total"].as_u64().unwrap();
1619         let roundup = total/200;
1620         let per = ((v+roundup)*100)/total;
1621         let info = format!(" ({} %)", per);
1622         Ok(format!("{} {:>8}", v, info))
1623     };
1624
1625     let options = default_table_format_options()
1626         .noheader(true)
1627         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1628         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1629         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1630
1631     let schema = &proxmox_backup::api2::admin::datastore::API_RETURN_SCHEMA_STATUS;
1632
1633     format_and_print_result_full(&mut data, schema, &output_format, &options);
1634
1635     Ok(Value::Null)
1636 }
1637
1638 // like get, but simply ignore errors and return Null instead
1639 async fn try_get(repo: &BackupRepository, url: &str) -> Value {
1640
1641     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
1642     let password = std::env::var(ENV_VAR_PBS_PASSWORD).ok();
1643
1644     let options = HttpClientOptions::new()
1645         .prefix(Some("proxmox-backup".to_string()))
1646         .password(password)
1647         .interactive(false)
1648         .fingerprint(fingerprint)
1649         .fingerprint_cache(true)
1650         .ticket_cache(true);
1651
1652     let client = match HttpClient::new(repo.host(), repo.user(), options) {
1653         Ok(v) => v,
1654         _ => return Value::Null,
1655     };
1656
1657     let mut resp = match client.get(url, None).await {
1658         Ok(v) => v,
1659         _ => return Value::Null,
1660     };
1661
1662     if let Some(map) = resp.as_object_mut() {
1663         if let Some(data) = map.remove("data") {
1664             return data;
1665         }
1666     }
1667     Value::Null
1668 }
1669
1670 fn complete_backup_group(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1671     proxmox_backup::tools::runtime::main(async { complete_backup_group_do(param).await })
1672 }
1673
1674 async fn complete_backup_group_do(param: &HashMap<String, String>) -> Vec<String> {
1675
1676     let mut result = vec![];
1677
1678     let repo = match extract_repository_from_map(param) {
1679         Some(v) => v,
1680         _ => return result,
1681     };
1682
1683     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
1684
1685     let data = try_get(&repo, &path).await;
1686
1687     if let Some(list) = data.as_array() {
1688         for item in list {
1689             if let (Some(backup_id), Some(backup_type)) =
1690                 (item["backup-id"].as_str(), item["backup-type"].as_str())
1691             {
1692                 result.push(format!("{}/{}", backup_type, backup_id));
1693             }
1694         }
1695     }
1696
1697     result
1698 }
1699
1700 pub fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1701     proxmox_backup::tools::runtime::main(async { complete_group_or_snapshot_do(arg, param).await })
1702 }
1703
1704 async fn complete_group_or_snapshot_do(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1705
1706     if arg.matches('/').count() < 2 {
1707         let groups = complete_backup_group_do(param).await;
1708         let mut result = vec![];
1709         for group in groups {
1710             result.push(group.to_string());
1711             result.push(format!("{}/", group));
1712         }
1713         return result;
1714     }
1715
1716     complete_backup_snapshot_do(param).await
1717 }
1718
1719 fn complete_backup_snapshot(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1720     proxmox_backup::tools::runtime::main(async { complete_backup_snapshot_do(param).await })
1721 }
1722
1723 async fn complete_backup_snapshot_do(param: &HashMap<String, String>) -> Vec<String> {
1724
1725     let mut result = vec![];
1726
1727     let repo = match extract_repository_from_map(param) {
1728         Some(v) => v,
1729         _ => return result,
1730     };
1731
1732     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
1733
1734     let data = try_get(&repo, &path).await;
1735
1736     if let Some(list) = data.as_array() {
1737         for item in list {
1738             if let (Some(backup_id), Some(backup_type), Some(backup_time)) =
1739                 (item["backup-id"].as_str(), item["backup-type"].as_str(), item["backup-time"].as_i64())
1740             {
1741                 let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
1742                 result.push(snapshot.relative_path().to_str().unwrap().to_owned());
1743             }
1744         }
1745     }
1746
1747     result
1748 }
1749
1750 fn complete_server_file_name(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1751     proxmox_backup::tools::runtime::main(async { complete_server_file_name_do(param).await })
1752 }
1753
1754 async fn complete_server_file_name_do(param: &HashMap<String, String>) -> Vec<String> {
1755
1756     let mut result = vec![];
1757
1758     let repo = match extract_repository_from_map(param) {
1759         Some(v) => v,
1760         _ => return result,
1761     };
1762
1763     let snapshot: BackupDir = match param.get("snapshot") {
1764         Some(path) => {
1765             match path.parse() {
1766                 Ok(v) => v,
1767                 _ => return result,
1768             }
1769         }
1770         _ => return result,
1771     };
1772
1773     let query = tools::json_object_to_query(json!({
1774         "backup-type": snapshot.group().backup_type(),
1775         "backup-id": snapshot.group().backup_id(),
1776         "backup-time": snapshot.backup_time().timestamp(),
1777     })).unwrap();
1778
1779     let path = format!("api2/json/admin/datastore/{}/files?{}", repo.store(), query);
1780
1781     let data = try_get(&repo, &path).await;
1782
1783     if let Some(list) = data.as_array() {
1784         for item in list {
1785             if let Some(filename) = item["filename"].as_str() {
1786                 result.push(filename.to_owned());
1787             }
1788         }
1789     }
1790
1791     result
1792 }
1793
1794 fn complete_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1795     complete_server_file_name(arg, param)
1796         .iter()
1797         .map(|v| tools::format::strip_server_file_expenstion(&v))
1798         .collect()
1799 }
1800
1801 pub fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1802     complete_server_file_name(arg, param)
1803         .iter()
1804         .filter_map(|v| {
1805             let name = tools::format::strip_server_file_expenstion(&v);
1806             if name.ends_with(".pxar") {
1807                 Some(name)
1808             } else {
1809                 None
1810             }
1811         })
1812         .collect()
1813 }
1814
1815 fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
1816
1817     let mut result = vec![];
1818
1819     let mut size = 64;
1820     loop {
1821         result.push(size.to_string());
1822         size *= 2;
1823         if size > 4096 { break; }
1824     }
1825
1826     result
1827 }
1828
1829 use proxmox_backup::client::RemoteChunkReader;
1830 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
1831 /// async use!
1832 ///
1833 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
1834 /// so that we can properly access it from multiple threads simultaneously while not issuing
1835 /// duplicate simultaneous reads over http.
1836 pub struct BufferedDynamicReadAt {
1837     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
1838 }
1839
1840 impl BufferedDynamicReadAt {
1841     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
1842         Self {
1843             inner: Mutex::new(inner),
1844         }
1845     }
1846 }
1847
1848 impl ReadAt for BufferedDynamicReadAt {
1849     fn start_read_at<'a>(
1850         self: Pin<&'a Self>,
1851         _cx: &mut Context,
1852         buf: &'a mut [u8],
1853         offset: u64,
1854     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1855         MaybeReady::Ready(tokio::task::block_in_place(move || {
1856             let mut reader = self.inner.lock().unwrap();
1857             reader.seek(SeekFrom::Start(offset))?;
1858             Ok(reader.read(buf)?)
1859         }))
1860     }
1861
1862     fn poll_complete<'a>(
1863         self: Pin<&'a Self>,
1864         _op: ReadAtOperation<'a>,
1865     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1866         panic!("LocalDynamicReadAt::start_read_at returned Pending");
1867     }
1868 }
1869
1870 fn main() {
1871
1872     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
1873         .arg_param(&["backupspec"])
1874         .completion_cb("repository", complete_repository)
1875         .completion_cb("backupspec", complete_backup_source)
1876         .completion_cb("keyfile", tools::complete_file_name)
1877         .completion_cb("chunk-size", complete_chunk_size);
1878
1879     let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
1880         .completion_cb("repository", complete_repository)
1881         .completion_cb("keyfile", tools::complete_file_name);
1882
1883     let upload_log_cmd_def = CliCommand::new(&API_METHOD_UPLOAD_LOG)
1884         .arg_param(&["snapshot", "logfile"])
1885         .completion_cb("snapshot", complete_backup_snapshot)
1886         .completion_cb("logfile", tools::complete_file_name)
1887         .completion_cb("keyfile", tools::complete_file_name)
1888         .completion_cb("repository", complete_repository);
1889
1890     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
1891         .completion_cb("repository", complete_repository);
1892
1893     let snapshots_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOTS)
1894         .arg_param(&["group"])
1895         .completion_cb("group", complete_backup_group)
1896         .completion_cb("repository", complete_repository);
1897
1898     let forget_cmd_def = CliCommand::new(&API_METHOD_FORGET_SNAPSHOTS)
1899         .arg_param(&["snapshot"])
1900         .completion_cb("repository", complete_repository)
1901         .completion_cb("snapshot", complete_backup_snapshot);
1902
1903     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
1904         .completion_cb("repository", complete_repository);
1905
1906     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
1907         .arg_param(&["snapshot", "archive-name", "target"])
1908         .completion_cb("repository", complete_repository)
1909         .completion_cb("snapshot", complete_group_or_snapshot)
1910         .completion_cb("archive-name", complete_archive_name)
1911         .completion_cb("target", tools::complete_file_name);
1912
1913     let files_cmd_def = CliCommand::new(&API_METHOD_LIST_SNAPSHOT_FILES)
1914         .arg_param(&["snapshot"])
1915         .completion_cb("repository", complete_repository)
1916         .completion_cb("snapshot", complete_backup_snapshot);
1917
1918     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
1919         .arg_param(&["group"])
1920         .completion_cb("group", complete_backup_group)
1921         .completion_cb("repository", complete_repository);
1922
1923     let status_cmd_def = CliCommand::new(&API_METHOD_STATUS)
1924         .completion_cb("repository", complete_repository);
1925
1926     let login_cmd_def = CliCommand::new(&API_METHOD_API_LOGIN)
1927         .completion_cb("repository", complete_repository);
1928
1929     let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
1930         .completion_cb("repository", complete_repository);
1931
1932     let version_cmd_def = CliCommand::new(&API_METHOD_API_VERSION)
1933         .completion_cb("repository", complete_repository);
1934
1935     let cmd_def = CliCommandMap::new()
1936         .insert("backup", backup_cmd_def)
1937         .insert("upload-log", upload_log_cmd_def)
1938         .insert("forget", forget_cmd_def)
1939         .insert("garbage-collect", garbage_collect_cmd_def)
1940         .insert("list", list_cmd_def)
1941         .insert("login", login_cmd_def)
1942         .insert("logout", logout_cmd_def)
1943         .insert("prune", prune_cmd_def)
1944         .insert("restore", restore_cmd_def)
1945         .insert("snapshots", snapshots_cmd_def)
1946         .insert("files", files_cmd_def)
1947         .insert("status", status_cmd_def)
1948         .insert("key", key::cli())
1949         .insert("mount", mount_cmd_def())
1950         .insert("catalog", catalog_mgmt_cli())
1951         .insert("task", task_mgmt_cli())
1952         .insert("version", version_cmd_def)
1953         .insert("benchmark", benchmark_cmd_def);
1954
1955     let rpcenv = CliEnvironment::new();
1956     run_cli_command(cmd_def, rpcenv, Some(|future| {
1957         proxmox_backup::tools::runtime::main(future)
1958     }));
1959 }