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     previous_manifest: Option<Arc<BackupManifest>>,
 284     dir_path: P,
 285     archive_name: &str,
 286     chunk_size: Option<usize>,
 287     device_set: Option<HashSet<u64>>,
 288     verbose: bool,
 289     skip_lost_and_found: bool,
 290     catalog: Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 291     exclude_pattern: Vec<MatchEntry>,
 292     entries_max: usize,
 293     compress: bool,
 294     encrypt: bool,
 295 ) -> Result<BackupStats, Error> {
 296
 297     let pxar_stream = PxarBackupStream::open(
 298         dir_path.as_ref(),
 299         device_set,
 300         verbose,
 301         skip_lost_and_found,
 302         catalog,
 303         exclude_pattern,
 304         entries_max,
 305     )?;
 306     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 307
 308     let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 309
 310     let stream = ReceiverStream::new(rx)
 311         .map_err(Error::from);
 312
 313     // spawn chunker inside a separate task so that it can run parallel
 314     tokio::spawn(async move {
 315         while let Some(v) = chunk_stream.next().await {
 316             let _ = tx.send(v).await;
 317         }
 318     });
 319
 320     let stats = client
 321         .upload_stream(previous_manifest, archive_name, stream, "dynamic", None, compress, encrypt)
 322         .await?;
 323
 324     Ok(stats)
 325 }
 326
 327 async fn backup_image<P: AsRef<Path>>(
 328     client: &BackupWriter,
 329     previous_manifest: Option<Arc<BackupManifest>>,
 330     image_path: P,
 331     archive_name: &str,
 332     image_size: u64,
 333     chunk_size: Option<usize>,
 334     compress: bool,
 335     encrypt: bool,
 336     _verbose: bool,
 337 ) -> Result<BackupStats, Error> {
 338
 339     let path = image_path.as_ref().to_owned();
 340
 341     let file = tokio::fs::File::open(path).await?;
 342
 343     let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 344         .map_err(Error::from);
 345
 346     let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
 347
 348     let stats = client
 349         .upload_stream(previous_manifest, archive_name, stream, "fixed", Some(image_size), compress, encrypt)
 350         .await?;
 351
 352     Ok(stats)
 353 }
 354
 355 #[api(
 356    input: {
 357         properties: {
 358             repository: {
 359                 schema: REPO_URL_SCHEMA,
 360                 optional: true,
 361             },
 362             "output-format": {
 363                 schema: OUTPUT_FORMAT,
 364                 optional: true,
 365             },
 366         }
 367    }
 368 )]
 369 /// List backup groups.
 370 async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 371
 372     let output_format = get_output_format(&param);
 373
 374     let repo = extract_repository_from_value(&param)?;
 375
 376     let client = connect(&repo)?;
 377
 378     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 379
 380     let mut result = client.get(&path, None).await?;
 381
 382     record_repository(&repo);
 383
 384     let render_group_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 385         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 386         let group = BackupGroup::new(item.backup_type, item.backup_id);
 387         Ok(group.group_path().to_str().unwrap().to_owned())
 388     };
 389
 390     let render_last_backup = |_v: &Value, record: &Value| -> Result<String, Error> {
 391         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 392         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.last_backup)?;
 393         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 394     };
 395
 396     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 397         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 398         Ok(tools::format::render_backup_file_list(&item.files))
 399     };
 400
 401     let options = default_table_format_options()
 402         .sortby("backup-type", false)
 403         .sortby("backup-id", false)
 404         .column(ColumnConfig::new("backup-id").renderer(render_group_path).header("group"))
 405         .column(
 406             ColumnConfig::new("last-backup")
 407                 .renderer(render_last_backup)
 408                 .header("last snapshot")
 409                 .right_align(false)
 410         )
 411         .column(ColumnConfig::new("backup-count"))
 412         .column(ColumnConfig::new("files").renderer(render_files));
 413
 414     let mut data: Value = result["data"].take();
 415
 416     let return_type = &proxmox_backup::api2::admin::datastore::API_METHOD_LIST_GROUPS.returns;
 417
 418     format_and_print_result_full(&mut data, return_type, &output_format, &options);
 419
 420     Ok(Value::Null)
 421 }
 422
 423 #[api(
 424    input: {
 425         properties: {
 426             repository: {
 427                 schema: REPO_URL_SCHEMA,
 428                 optional: true,
 429             },
 430             group: {
 431                 type: String,
 432                 description: "Backup group.",
 433             },
 434             "new-owner": {
 435                 type: Authid,
 436             },
 437         }
 438    }
 439 )]
 440 /// Change owner of a backup group
 441 async fn change_backup_owner(group: String, mut param: Value) -> Result<(), Error> {
 442
 443     let repo = extract_repository_from_value(&param)?;
 444
 445     let mut client = connect(&repo)?;
 446
 447     param.as_object_mut().unwrap().remove("repository");
 448
 449     let group: BackupGroup = group.parse()?;
 450
 451     param["backup-type"] = group.backup_type().into();
 452     param["backup-id"] = group.backup_id().into();
 453
 454     let path = format!("api2/json/admin/datastore/{}/change-owner", repo.store());
 455     client.post(&path, Some(param)).await?;
 456
 457     record_repository(&repo);
 458
 459     Ok(())
 460 }
 461
 462 #[api(
 463    input: {
 464         properties: {
 465             repository: {
 466                 schema: REPO_URL_SCHEMA,
 467                 optional: true,
 468             },
 469         }
 470    }
 471 )]
 472 /// Try to login. If successful, store ticket.
 473 async fn api_login(param: Value) -> Result<Value, Error> {
 474
 475     let repo = extract_repository_from_value(&param)?;
 476
 477     let client = connect(&repo)?;
 478     client.login().await?;
 479
 480     record_repository(&repo);
 481
 482     Ok(Value::Null)
 483 }
 484
 485 #[api(
 486    input: {
 487         properties: {
 488             repository: {
 489                 schema: REPO_URL_SCHEMA,
 490                 optional: true,
 491             },
 492         }
 493    }
 494 )]
 495 /// Logout (delete stored ticket).
 496 fn api_logout(param: Value) -> Result<Value, Error> {
 497
 498     let repo = extract_repository_from_value(&param)?;
 499
 500     delete_ticket_info("proxmox-backup", repo.host(), repo.user())?;
 501
 502     Ok(Value::Null)
 503 }
 504
 505 #[api(
 506    input: {
 507         properties: {
 508             repository: {
 509                 schema: REPO_URL_SCHEMA,
 510                 optional: true,
 511             },
 512             "output-format": {
 513                 schema: OUTPUT_FORMAT,
 514                 optional: true,
 515             },
 516         }
 517    }
 518 )]
 519 /// Show client and optional server version
 520 async fn api_version(param: Value) -> Result<(), Error> {
 521
 522     let output_format = get_output_format(&param);
 523
 524     let mut version_info = json!({
 525         "client": {
 526             "version": version::PROXMOX_PKG_VERSION,
 527             "release": version::PROXMOX_PKG_RELEASE,
 528             "repoid": version::PROXMOX_PKG_REPOID,
 529         }
 530     });
 531
 532     let repo = extract_repository_from_value(&param);
 533     if let Ok(repo) = repo {
 534         let client = connect(&repo)?;
 535
 536         match client.get("api2/json/version", None).await {
 537             Ok(mut result) => version_info["server"] = result["data"].take(),
 538             Err(e) => eprintln!("could not connect to server - {}", e),
 539         }
 540     }
 541     if output_format == "text" {
 542         println!("client version: {}.{}", version::PROXMOX_PKG_VERSION, version::PROXMOX_PKG_RELEASE);
 543         if let Some(server) = version_info["server"].as_object() {
 544             let server_version = server["version"].as_str().unwrap();
 545             let server_release = server["release"].as_str().unwrap();
 546             println!("server version: {}.{}", server_version, server_release);
 547         }
 548     } else {
 549         format_and_print_result(&version_info, &output_format);
 550     }
 551
 552     Ok(())
 553 }
 554
 555 #[api(
 556     input: {
 557         properties: {
 558             repository: {
 559                 schema: REPO_URL_SCHEMA,
 560                 optional: true,
 561             },
 562             "output-format": {
 563                 schema: OUTPUT_FORMAT,
 564                 optional: true,
 565             },
 566         },
 567     },
 568 )]
 569 /// Start garbage collection for a specific repository.
 570 async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 571
 572     let repo = extract_repository_from_value(&param)?;
 573
 574     let output_format = get_output_format(&param);
 575
 576     let mut client = connect(&repo)?;
 577
 578     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 579
 580     let result = client.post(&path, None).await?;
 581
 582     record_repository(&repo);
 583
 584     view_task_result(client, result, &output_format).await?;
 585
 586     Ok(Value::Null)
 587 }
 588
 589 fn spawn_catalog_upload(
 590     client: Arc<BackupWriter>,
 591     encrypt: bool,
 592 ) -> Result<
 593         (
 594             Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
 595             tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>
 596         ), Error>
 597 {
 598     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
 599     let catalog_stream = crate::tools::StdChannelStream(catalog_rx);
 600     let catalog_chunk_size = 512*1024;
 601     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 602
 603     let catalog = Arc::new(Mutex::new(CatalogWriter::new(crate::tools::StdChannelWriter::new(catalog_tx))?));
 604
 605     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 606
 607     tokio::spawn(async move {
 608         let catalog_upload_result = client
 609             .upload_stream(None, CATALOG_NAME, catalog_chunk_stream, "dynamic", None, true, encrypt)
 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((catalog, 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                     key_config.fingerprint = Some(fingerprint);
 930
 931                     let enc_key = rsa_encrypt_key_config(rsa, &key_config)?;
 932                     println!("Master key '{:?}'", path);
 933
 934                     (Some(Arc::new(crypt_config)), Some(enc_key))
 935                 }
 936                 _ => (Some(Arc::new(crypt_config)), None),
 937             }
 938         }
 939     };
 940
 941     let client = BackupWriter::start(
 942         client,
 943         crypt_config.clone(),
 944         repo.store(),
 945         backup_type,
 946         &backup_id,
 947         backup_time,
 948         verbose,
 949         false
 950     ).await?;
 951
 952     let download_previous_manifest = match client.previous_backup_time().await {
 953         Ok(Some(backup_time)) => {
 954             println!(
 955                 "Downloading previous manifest ({})",
 956                 strftime_local("%c", backup_time)?
 957             );
 958             true
 959         }
 960         Ok(None) => {
 961             println!("No previous manifest available.");
 962             false
 963         }
 964         Err(_) => {
 965             // Fallback for outdated server, TODO remove/bubble up with 2.0
 966             true
 967         }
 968     };
 969
 970     let previous_manifest = if download_previous_manifest {
 971         match client.download_previous_manifest().await {
 972             Ok(previous_manifest) => {
 973                 match previous_manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref)) {
 974                     Ok(()) => Some(Arc::new(previous_manifest)),
 975                     Err(err) => {
 976                         println!("Couldn't re-use previous manifest - {}", err);
 977                         None
 978                     }
 979                 }
 980             }
 981             Err(err) => {
 982                 println!("Couldn't download previous manifest - {}", err);
 983                 None
 984             }
 985         }
 986     } else {
 987         None
 988     };
 989
 990     let snapshot = BackupDir::new(backup_type, backup_id, backup_time)?;
 991     let mut manifest = BackupManifest::new(snapshot);
 992
 993     let mut catalog = None;
 994     let mut catalog_result_tx = None;
 995
 996     for (backup_type, filename, target, size) in upload_list {
 997         match backup_type {
 998             BackupSpecificationType::CONFIG => {
 999                 println!("Upload config file '{}' to '{}' as {}", filename, repo, target);
1000                 let stats = client
1001                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1002                     .await?;
1003                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1004             }
1005             BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
1006                 println!("Upload log file '{}' to '{}' as {}", filename, repo, target);
1007                 let stats = client
1008                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1009                     .await?;
1010                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1011             }
1012             BackupSpecificationType::PXAR => {
1013                 // start catalog upload on first use
1014                 if catalog.is_none() {
1015                     let (cat, res) = spawn_catalog_upload(client.clone(), crypt_mode == CryptMode::Encrypt)?;
1016                     catalog = Some(cat);
1017                     catalog_result_tx = Some(res);
1018                 }
1019                 let catalog = catalog.as_ref().unwrap();
1020
1021                 println!("Upload directory '{}' to '{}' as {}", filename, repo, target);
1022                 catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
1023                 let stats = backup_directory(
1024                     &client,
1025                     previous_manifest.clone(),
1026                     &filename,
1027                     &target,
1028                     chunk_size_opt,
1029                     devices.clone(),
1030                     verbose,
1031                     skip_lost_and_found,
1032                     catalog.clone(),
1033                     pattern_list.clone(),
1034                     entries_max as usize,
1035                     true,
1036                     crypt_mode == CryptMode::Encrypt,
1037                 ).await?;
1038                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1039                 catalog.lock().unwrap().end_directory()?;
1040             }
1041             BackupSpecificationType::IMAGE => {
1042                 println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
1043                 let stats = backup_image(
1044                     &client,
1045                     previous_manifest.clone(),
1046                      &filename,
1047                     &target,
1048                     size,
1049                     chunk_size_opt,
1050                     true,
1051                     crypt_mode == CryptMode::Encrypt,
1052                     verbose,
1053                 ).await?;
1054                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1055             }
1056         }
1057     }
1058
1059     // finalize and upload catalog
1060     if let Some(catalog) = catalog {
1061         let mutex = Arc::try_unwrap(catalog)
1062             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
1063         let mut catalog = mutex.into_inner().unwrap();
1064
1065         catalog.finish()?;
1066
1067         drop(catalog); // close upload stream
1068
1069         if let Some(catalog_result_rx) = catalog_result_tx {
1070             let stats = catalog_result_rx.await??;
1071             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypt_mode)?;
1072         }
1073     }
1074
1075     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
1076         let target = ENCRYPTED_KEY_BLOB_NAME;
1077         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
1078         let stats = client
1079             .upload_blob_from_data(rsa_encrypted_key, target, false, false)
1080             .await?;
1081         manifest.add_file(target.to_string(), stats.size, stats.csum, crypt_mode)?;
1082
1083     }
1084     // create manifest (index.json)
1085     // manifests are never encrypted, but include a signature
1086     let manifest = manifest.to_string(crypt_config.as_ref().map(Arc::as_ref))
1087         .map_err(|err| format_err!("unable to format manifest - {}", err))?;
1088
1089
1090     if verbose { println!("Upload index.json to '{}'", repo) };
1091     client
1092         .upload_blob_from_data(manifest.into_bytes(), MANIFEST_BLOB_NAME, true, false)
1093         .await?;
1094
1095     client.finish().await?;
1096
1097     let end_time = std::time::Instant::now();
1098     let elapsed = end_time.duration_since(start_time);
1099     println!("Duration: {:.2}s", elapsed.as_secs_f64());
1100
1101     println!("End Time: {}", strftime_local("%c", epoch_i64())?);
1102
1103     Ok(Value::Null)
1104 }
1105
1106 fn complete_backup_source(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1107
1108     let mut result = vec![];
1109
1110     let data: Vec<&str> = arg.splitn(2, ':').collect();
1111
1112     if data.len() != 2 {
1113         result.push(String::from("root.pxar:/"));
1114         result.push(String::from("etc.pxar:/etc"));
1115         return result;
1116     }
1117
1118     let files = tools::complete_file_name(data[1], param);
1119
1120     for file in files {
1121         result.push(format!("{}:{}", data[0], file));
1122     }
1123
1124     result
1125 }
1126
1127 async fn dump_image<W: Write>(
1128     client: Arc<BackupReader>,
1129     crypt_config: Option<Arc<CryptConfig>>,
1130     crypt_mode: CryptMode,
1131     index: FixedIndexReader,
1132     mut writer: W,
1133     verbose: bool,
1134 ) -> Result<(), Error> {
1135
1136     let most_used = index.find_most_used_chunks(8);
1137
1138     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, crypt_mode, most_used);
1139
1140     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
1141     // and thus slows down reading. Instead, directly use RemoteChunkReader
1142     let mut per = 0;
1143     let mut bytes = 0;
1144     let start_time = std::time::Instant::now();
1145
1146     for pos in 0..index.index_count() {
1147         let digest = index.index_digest(pos).unwrap();
1148         let raw_data = chunk_reader.read_chunk(&digest).await?;
1149         writer.write_all(&raw_data)?;
1150         bytes += raw_data.len();
1151         if verbose {
1152             let next_per = ((pos+1)*100)/index.index_count();
1153             if per != next_per {
1154                 eprintln!("progress {}% (read {} bytes, duration {} sec)",
1155                           next_per, bytes, start_time.elapsed().as_secs());
1156                 per = next_per;
1157             }
1158         }
1159     }
1160
1161     let end_time = std::time::Instant::now();
1162     let elapsed = end_time.duration_since(start_time);
1163     eprintln!("restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1164               bytes,
1165               elapsed.as_secs_f64(),
1166               bytes as f64/(1024.0*1024.0*elapsed.as_secs_f64())
1167     );
1168
1169
1170     Ok(())
1171 }
1172
1173 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1174     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1175         (name.into(), archive_type(name).unwrap())
1176     } else if name.ends_with(".pxar") {
1177         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1178     } else if name.ends_with(".img") {
1179         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1180     } else {
1181         (format!("{}.blob", name), ArchiveType::Blob)
1182     }
1183 }
1184
1185 #[api(
1186    input: {
1187        properties: {
1188            repository: {
1189                schema: REPO_URL_SCHEMA,
1190                optional: true,
1191            },
1192            snapshot: {
1193                type: String,
1194                description: "Group/Snapshot path.",
1195            },
1196            "archive-name": {
1197                description: "Backup archive name.",
1198                type: String,
1199            },
1200            target: {
1201                type: String,
1202                description: r###"Target directory path. Use '-' to write to standard output.
1203
1204 We do not extraxt '.pxar' archives when writing to standard output.
1205
1206 "###
1207            },
1208            "allow-existing-dirs": {
1209                type: Boolean,
1210                description: "Do not fail if directories already exists.",
1211                optional: true,
1212            },
1213            keyfile: {
1214                schema: KEYFILE_SCHEMA,
1215                optional: true,
1216            },
1217            "keyfd": {
1218                schema: KEYFD_SCHEMA,
1219                optional: true,
1220            },
1221            "crypt-mode": {
1222                type: CryptMode,
1223                optional: true,
1224            },
1225        }
1226    }
1227 )]
1228 /// Restore backup repository.
1229 async fn restore(param: Value) -> Result<Value, Error> {
1230     let repo = extract_repository_from_value(&param)?;
1231
1232     let verbose = param["verbose"].as_bool().unwrap_or(false);
1233
1234     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1235
1236     let archive_name = tools::required_string_param(&param, "archive-name")?;
1237
1238     let client = connect(&repo)?;
1239
1240     record_repository(&repo);
1241
1242     let path = tools::required_string_param(&param, "snapshot")?;
1243
1244     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
1245         let group: BackupGroup = path.parse()?;
1246         api_datastore_latest_snapshot(&client, repo.store(), group).await?
1247     } else {
1248         let snapshot: BackupDir = path.parse()?;
1249         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
1250     };
1251
1252     let target = tools::required_string_param(&param, "target")?;
1253     let target = if target == "-" { None } else { Some(target) };
1254
1255     let (keydata, _crypt_mode) = keyfile_parameters(&param)?;
1256
1257     let crypt_config = match keydata {
1258         None => None,
1259         Some(key) => {
1260             let (key, _, fingerprint) = decrypt_key(&key, &key::get_encryption_key_password)?;
1261             eprintln!("Encryption key fingerprint: '{}'", fingerprint);
1262             Some(Arc::new(CryptConfig::new(key)?))
1263         }
1264     };
1265
1266     let client = BackupReader::start(
1267         client,
1268         crypt_config.clone(),
1269         repo.store(),
1270         &backup_type,
1271         &backup_id,
1272         backup_time,
1273         true,
1274     ).await?;
1275
1276     let (archive_name, archive_type) = parse_archive_type(archive_name);
1277
1278     let (manifest, backup_index_data) = client.download_manifest().await?;
1279
1280     if archive_name == ENCRYPTED_KEY_BLOB_NAME && crypt_config.is_none() {
1281         eprintln!("Restoring encrypted key blob without original key - skipping manifest fingerprint check!")
1282     } else {
1283         manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref))?;
1284     }
1285
1286     if archive_name == MANIFEST_BLOB_NAME {
1287         if let Some(target) = target {
1288             replace_file(target, &backup_index_data, CreateOptions::new())?;
1289         } else {
1290             let stdout = std::io::stdout();
1291             let mut writer = stdout.lock();
1292             writer.write_all(&backup_index_data)
1293                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1294         }
1295
1296         return Ok(Value::Null);
1297     }
1298
1299     let file_info = manifest.lookup_file_info(&archive_name)?;
1300
1301     if archive_type == ArchiveType::Blob {
1302
1303         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1304
1305         if let Some(target) = target {
1306            let mut writer = std::fs::OpenOptions::new()
1307                 .write(true)
1308                 .create(true)
1309                 .create_new(true)
1310                 .open(target)
1311                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?;
1312             std::io::copy(&mut reader, &mut writer)?;
1313         } else {
1314             let stdout = std::io::stdout();
1315             let mut writer = stdout.lock();
1316             std::io::copy(&mut reader, &mut writer)
1317                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1318         }
1319
1320     } else if archive_type == ArchiveType::DynamicIndex {
1321
1322         let index = client.download_dynamic_index(&manifest, &archive_name).await?;
1323
1324         let most_used = index.find_most_used_chunks(8);
1325
1326         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, file_info.chunk_crypt_mode(), most_used);
1327
1328         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1329
1330         if let Some(target) = target {
1331             proxmox_backup::pxar::extract_archive(
1332                 pxar::decoder::Decoder::from_std(reader)?,
1333                 Path::new(target),
1334                 &[],
1335                 true,
1336                 proxmox_backup::pxar::Flags::DEFAULT,
1337                 allow_existing_dirs,
1338                 |path| {
1339                     if verbose {
1340                         println!("{:?}", path);
1341                     }
1342                 },
1343                 None,
1344             )
1345             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1346         } else {
1347             let mut writer = std::fs::OpenOptions::new()
1348                 .write(true)
1349                 .open("/dev/stdout")
1350                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1351
1352             std::io::copy(&mut reader, &mut writer)
1353                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1354         }
1355     } else if archive_type == ArchiveType::FixedIndex {
1356
1357         let index = client.download_fixed_index(&manifest, &archive_name).await?;
1358
1359         let mut writer = if let Some(target) = target {
1360             std::fs::OpenOptions::new()
1361                 .write(true)
1362                 .create(true)
1363                 .create_new(true)
1364                 .open(target)
1365                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1366         } else {
1367             std::fs::OpenOptions::new()
1368                 .write(true)
1369                 .open("/dev/stdout")
1370                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1371         };
1372
1373         dump_image(client.clone(), crypt_config.clone(), file_info.chunk_crypt_mode(), index, &mut writer, verbose).await?;
1374     }
1375
1376     Ok(Value::Null)
1377 }
1378
1379 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
1380     &ApiHandler::Async(&prune),
1381     &ObjectSchema::new(
1382         "Prune a backup repository.",
1383         &proxmox_backup::add_common_prune_prameters!([
1384             ("dry-run", true, &BooleanSchema::new(
1385                 "Just show what prune would do, but do not delete anything.")
1386              .schema()),
1387             ("group", false, &StringSchema::new("Backup group.").schema()),
1388         ], [
1389             ("output-format", true, &OUTPUT_FORMAT),
1390             (
1391                 "quiet",
1392                 true,
1393                 &BooleanSchema::new("Minimal output - only show removals.")
1394                     .schema()
1395             ),
1396             ("repository", true, &REPO_URL_SCHEMA),
1397         ])
1398     )
1399 );
1400
1401 fn prune<'a>(
1402     param: Value,
1403     _info: &ApiMethod,
1404     _rpcenv: &'a mut dyn RpcEnvironment,
1405 ) -> proxmox::api::ApiFuture<'a> {
1406     async move {
1407         prune_async(param).await
1408     }.boxed()
1409 }
1410
1411 async fn prune_async(mut param: Value) -> Result<Value, Error> {
1412     let repo = extract_repository_from_value(&param)?;
1413
1414     let mut client = connect(&repo)?;
1415
1416     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1417
1418     let group = tools::required_string_param(&param, "group")?;
1419     let group: BackupGroup = group.parse()?;
1420
1421     let output_format = get_output_format(&param);
1422
1423     let quiet = param["quiet"].as_bool().unwrap_or(false);
1424
1425     param.as_object_mut().unwrap().remove("repository");
1426     param.as_object_mut().unwrap().remove("group");
1427     param.as_object_mut().unwrap().remove("output-format");
1428     param.as_object_mut().unwrap().remove("quiet");
1429
1430     param["backup-type"] = group.backup_type().into();
1431     param["backup-id"] = group.backup_id().into();
1432
1433     let mut result = client.post(&path, Some(param)).await?;
1434
1435     record_repository(&repo);
1436
1437     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1438         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1439         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time)?;
1440         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
1441     };
1442
1443     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1444         Ok(match v.as_bool() {
1445             Some(true) => "keep",
1446             Some(false) => "remove",
1447             None => "unknown",
1448         }.to_string())
1449     };
1450
1451     let options = default_table_format_options()
1452         .sortby("backup-type", false)
1453         .sortby("backup-id", false)
1454         .sortby("backup-time", false)
1455         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
1456         .column(ColumnConfig::new("backup-time").renderer(tools::format::render_epoch).header("date"))
1457         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
1458         ;
1459
1460     let return_type = &proxmox_backup::api2::admin::datastore::API_METHOD_PRUNE.returns;
1461
1462     let mut data = result["data"].take();
1463
1464     if quiet {
1465         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
1466             item["keep"].as_bool() == Some(false)
1467         }).cloned().collect();
1468         data = list.into();
1469     }
1470
1471     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1472
1473     Ok(Value::Null)
1474 }
1475
1476 #[api(
1477    input: {
1478        properties: {
1479            repository: {
1480                schema: REPO_URL_SCHEMA,
1481                optional: true,
1482            },
1483            "output-format": {
1484                schema: OUTPUT_FORMAT,
1485                optional: true,
1486            },
1487        }
1488    },
1489     returns: {
1490         type: StorageStatus,
1491     },
1492 )]
1493 /// Get repository status.
1494 async fn status(param: Value) -> Result<Value, Error> {
1495
1496     let repo = extract_repository_from_value(&param)?;
1497
1498     let output_format = get_output_format(&param);
1499
1500     let client = connect(&repo)?;
1501
1502     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1503
1504     let mut result = client.get(&path, None).await?;
1505     let mut data = result["data"].take();
1506
1507     record_repository(&repo);
1508
1509     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1510         let v = v.as_u64().unwrap();
1511         let total = record["total"].as_u64().unwrap();
1512         let roundup = total/200;
1513         let per = ((v+roundup)*100)/total;
1514         let info = format!(" ({} %)", per);
1515         Ok(format!("{} {:>8}", v, info))
1516     };
1517
1518     let options = default_table_format_options()
1519         .noheader(true)
1520         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1521         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1522         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1523
1524     let return_type = &API_METHOD_STATUS.returns;
1525
1526     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1527
1528     Ok(Value::Null)
1529 }
1530
1531 // like get, but simply ignore errors and return Null instead
1532 async fn try_get(repo: &BackupRepository, url: &str) -> Value {
1533
1534     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
1535     let password = std::env::var(ENV_VAR_PBS_PASSWORD).ok();
1536
1537     let options = HttpClientOptions::new()
1538         .prefix(Some("proxmox-backup".to_string()))
1539         .password(password)
1540         .interactive(false)
1541         .fingerprint(fingerprint)
1542         .fingerprint_cache(true)
1543         .ticket_cache(true);
1544
1545     let client = match HttpClient::new(repo.host(), repo.port(), repo.auth_id(), options) {
1546         Ok(v) => v,
1547         _ => return Value::Null,
1548     };
1549
1550     let mut resp = match client.get(url, None).await {
1551         Ok(v) => v,
1552         _ => return Value::Null,
1553     };
1554
1555     if let Some(map) = resp.as_object_mut() {
1556         if let Some(data) = map.remove("data") {
1557             return data;
1558         }
1559     }
1560     Value::Null
1561 }
1562
1563 fn complete_backup_group(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1564     proxmox_backup::tools::runtime::main(async { complete_backup_group_do(param).await })
1565 }
1566
1567 async fn complete_backup_group_do(param: &HashMap<String, String>) -> Vec<String> {
1568
1569     let mut result = vec![];
1570
1571     let repo = match extract_repository_from_map(param) {
1572         Some(v) => v,
1573         _ => return result,
1574     };
1575
1576     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
1577
1578     let data = try_get(&repo, &path).await;
1579
1580     if let Some(list) = data.as_array() {
1581         for item in list {
1582             if let (Some(backup_id), Some(backup_type)) =
1583                 (item["backup-id"].as_str(), item["backup-type"].as_str())
1584             {
1585                 result.push(format!("{}/{}", backup_type, backup_id));
1586             }
1587         }
1588     }
1589
1590     result
1591 }
1592
1593 pub fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1594     proxmox_backup::tools::runtime::main(async { complete_group_or_snapshot_do(arg, param).await })
1595 }
1596
1597 async fn complete_group_or_snapshot_do(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1598
1599     if arg.matches('/').count() < 2 {
1600         let groups = complete_backup_group_do(param).await;
1601         let mut result = vec![];
1602         for group in groups {
1603             result.push(group.to_string());
1604             result.push(format!("{}/", group));
1605         }
1606         return result;
1607     }
1608
1609     complete_backup_snapshot_do(param).await
1610 }
1611
1612 fn complete_backup_snapshot(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1613     proxmox_backup::tools::runtime::main(async { complete_backup_snapshot_do(param).await })
1614 }
1615
1616 async fn complete_backup_snapshot_do(param: &HashMap<String, String>) -> Vec<String> {
1617
1618     let mut result = vec![];
1619
1620     let repo = match extract_repository_from_map(param) {
1621         Some(v) => v,
1622         _ => return result,
1623     };
1624
1625     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
1626
1627     let data = try_get(&repo, &path).await;
1628
1629     if let Some(list) = data.as_array() {
1630         for item in list {
1631             if let (Some(backup_id), Some(backup_type), Some(backup_time)) =
1632                 (item["backup-id"].as_str(), item["backup-type"].as_str(), item["backup-time"].as_i64())
1633             {
1634                 if let Ok(snapshot) = BackupDir::new(backup_type, backup_id, backup_time) {
1635                     result.push(snapshot.relative_path().to_str().unwrap().to_owned());
1636                 }
1637             }
1638         }
1639     }
1640
1641     result
1642 }
1643
1644 fn complete_server_file_name(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1645     proxmox_backup::tools::runtime::main(async { complete_server_file_name_do(param).await })
1646 }
1647
1648 async fn complete_server_file_name_do(param: &HashMap<String, String>) -> Vec<String> {
1649
1650     let mut result = vec![];
1651
1652     let repo = match extract_repository_from_map(param) {
1653         Some(v) => v,
1654         _ => return result,
1655     };
1656
1657     let snapshot: BackupDir = match param.get("snapshot") {
1658         Some(path) => {
1659             match path.parse() {
1660                 Ok(v) => v,
1661                 _ => return result,
1662             }
1663         }
1664         _ => return result,
1665     };
1666
1667     let query = tools::json_object_to_query(json!({
1668         "backup-type": snapshot.group().backup_type(),
1669         "backup-id": snapshot.group().backup_id(),
1670         "backup-time": snapshot.backup_time(),
1671     })).unwrap();
1672
1673     let path = format!("api2/json/admin/datastore/{}/files?{}", repo.store(), query);
1674
1675     let data = try_get(&repo, &path).await;
1676
1677     if let Some(list) = data.as_array() {
1678         for item in list {
1679             if let Some(filename) = item["filename"].as_str() {
1680                 result.push(filename.to_owned());
1681             }
1682         }
1683     }
1684
1685     result
1686 }
1687
1688 fn complete_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1689     complete_server_file_name(arg, param)
1690         .iter()
1691         .map(|v| tools::format::strip_server_file_extension(&v))
1692         .collect()
1693 }
1694
1695 pub fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1696     complete_server_file_name(arg, param)
1697         .iter()
1698         .filter_map(|name| {
1699             if name.ends_with(".pxar.didx") {
1700                 Some(tools::format::strip_server_file_extension(name))
1701             } else {
1702                 None
1703             }
1704         })
1705         .collect()
1706 }
1707
1708 pub fn complete_img_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1709     complete_server_file_name(arg, param)
1710         .iter()
1711         .filter_map(|name| {
1712             if name.ends_with(".img.fidx") {
1713                 Some(tools::format::strip_server_file_extension(name))
1714             } else {
1715                 None
1716             }
1717         })
1718         .collect()
1719 }
1720
1721 fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
1722
1723     let mut result = vec![];
1724
1725     let mut size = 64;
1726     loop {
1727         result.push(size.to_string());
1728         size *= 2;
1729         if size > 4096 { break; }
1730     }
1731
1732     result
1733 }
1734
1735 fn complete_auth_id(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1736     proxmox_backup::tools::runtime::main(async { complete_auth_id_do(param).await })
1737 }
1738
1739 async fn complete_auth_id_do(param: &HashMap<String, String>) -> Vec<String> {
1740
1741     let mut result = vec![];
1742
1743     let repo = match extract_repository_from_map(param) {
1744         Some(v) => v,
1745         _ => return result,
1746     };
1747
1748     let data = try_get(&repo, "api2/json/access/users?include_tokens=true").await;
1749
1750     if let Ok(parsed) = serde_json::from_value::<Vec<UserWithTokens>>(data) {
1751         for user in parsed {
1752             result.push(user.userid.to_string());
1753             for token in user.tokens {
1754                 result.push(token.tokenid.to_string());
1755             }
1756         }
1757     };
1758
1759     result
1760 }
1761
1762 use proxmox_backup::client::RemoteChunkReader;
1763 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
1764 /// async use!
1765 ///
1766 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
1767 /// so that we can properly access it from multiple threads simultaneously while not issuing
1768 /// duplicate simultaneous reads over http.
1769 pub struct BufferedDynamicReadAt {
1770     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
1771 }
1772
1773 impl BufferedDynamicReadAt {
1774     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
1775         Self {
1776             inner: Mutex::new(inner),
1777         }
1778     }
1779 }
1780
1781 impl ReadAt for BufferedDynamicReadAt {
1782     fn start_read_at<'a>(
1783         self: Pin<&'a Self>,
1784         _cx: &mut Context,
1785         buf: &'a mut [u8],
1786         offset: u64,
1787     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1788         MaybeReady::Ready(tokio::task::block_in_place(move || {
1789             let mut reader = self.inner.lock().unwrap();
1790             reader.seek(SeekFrom::Start(offset))?;
1791             Ok(reader.read(buf)?)
1792         }))
1793     }
1794
1795     fn poll_complete<'a>(
1796         self: Pin<&'a Self>,
1797         _op: ReadAtOperation<'a>,
1798     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1799         panic!("LocalDynamicReadAt::start_read_at returned Pending");
1800     }
1801 }
1802
1803 fn main() {
1804
1805     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
1806         .arg_param(&["backupspec"])
1807         .completion_cb("repository", complete_repository)
1808         .completion_cb("backupspec", complete_backup_source)
1809         .completion_cb("keyfile", tools::complete_file_name)
1810         .completion_cb("chunk-size", complete_chunk_size);
1811
1812     let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
1813         .completion_cb("repository", complete_repository)
1814         .completion_cb("keyfile", tools::complete_file_name);
1815
1816     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
1817         .completion_cb("repository", complete_repository);
1818
1819     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
1820         .completion_cb("repository", complete_repository);
1821
1822     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
1823         .arg_param(&["snapshot", "archive-name", "target"])
1824         .completion_cb("repository", complete_repository)
1825         .completion_cb("snapshot", complete_group_or_snapshot)
1826         .completion_cb("archive-name", complete_archive_name)
1827         .completion_cb("target", tools::complete_file_name);
1828
1829     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
1830         .arg_param(&["group"])
1831         .completion_cb("group", complete_backup_group)
1832         .completion_cb("repository", complete_repository);
1833
1834     let status_cmd_def = CliCommand::new(&API_METHOD_STATUS)
1835         .completion_cb("repository", complete_repository);
1836
1837     let login_cmd_def = CliCommand::new(&API_METHOD_API_LOGIN)
1838         .completion_cb("repository", complete_repository);
1839
1840     let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
1841         .completion_cb("repository", complete_repository);
1842
1843     let version_cmd_def = CliCommand::new(&API_METHOD_API_VERSION)
1844         .completion_cb("repository", complete_repository);
1845
1846     let change_owner_cmd_def = CliCommand::new(&API_METHOD_CHANGE_BACKUP_OWNER)
1847         .arg_param(&["group", "new-owner"])
1848         .completion_cb("group", complete_backup_group)
1849         .completion_cb("new-owner",  complete_auth_id)
1850         .completion_cb("repository", complete_repository);
1851
1852     let cmd_def = CliCommandMap::new()
1853         .insert("backup", backup_cmd_def)
1854         .insert("garbage-collect", garbage_collect_cmd_def)
1855         .insert("list", list_cmd_def)
1856         .insert("login", login_cmd_def)
1857         .insert("logout", logout_cmd_def)
1858         .insert("prune", prune_cmd_def)
1859         .insert("restore", restore_cmd_def)
1860         .insert("snapshot", snapshot_mgtm_cli())
1861         .insert("status", status_cmd_def)
1862         .insert("key", key::cli())
1863         .insert("mount", mount_cmd_def())
1864         .insert("map", map_cmd_def())
1865         .insert("unmap", unmap_cmd_def())
1866         .insert("catalog", catalog_mgmt_cli())
1867         .insert("task", task_mgmt_cli())
1868         .insert("version", version_cmd_def)
1869         .insert("benchmark", benchmark_cmd_def)
1870         .insert("change-owner", change_owner_cmd_def)
1871
1872         .alias(&["files"], &["snapshot", "files"])
1873         .alias(&["forget"], &["snapshot", "forget"])
1874         .alias(&["upload-log"], &["snapshot", "upload-log"])
1875         .alias(&["snapshots"], &["snapshot", "list"])
1876         ;
1877
1878     let rpcenv = CliEnvironment::new();
1879     run_cli_command(cmd_def, rpcenv, Some(|future| {
1880         proxmox_backup::tools::runtime::main(future)
1881     }));
1882 }