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