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                     let key_config = KeyConfig {
 927                         kdf: None,
 928                         created,
 929                         modified: proxmox::tools::time::epoch_i64(),
 930                         data: key.to_vec(),
 931                         fingerprint: Some(fingerprint),
 932                     };
 933                     let enc_key = rsa_encrypt_key_config(rsa, &key_config)?;
 934                     println!("Master key '{:?}'", path);
 935
 936                     (Some(Arc::new(crypt_config)), Some(enc_key))
 937                 }
 938                 _ => (Some(Arc::new(crypt_config)), None),
 939             }
 940         }
 941     };
 942
 943     let client = BackupWriter::start(
 944         client,
 945         crypt_config.clone(),
 946         repo.store(),
 947         backup_type,
 948         &backup_id,
 949         backup_time,
 950         verbose,
 951         false
 952     ).await?;
 953
 954     let download_previous_manifest = match client.previous_backup_time().await {
 955         Ok(Some(backup_time)) => {
 956             println!(
 957                 "Downloading previous manifest ({})",
 958                 strftime_local("%c", backup_time)?
 959             );
 960             true
 961         }
 962         Ok(None) => {
 963             println!("No previous manifest available.");
 964             false
 965         }
 966         Err(_) => {
 967             // Fallback for outdated server, TODO remove/bubble up with 2.0
 968             true
 969         }
 970     };
 971
 972     let previous_manifest = if download_previous_manifest {
 973         match client.download_previous_manifest().await {
 974             Ok(previous_manifest) => {
 975                 match previous_manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref)) {
 976                     Ok(()) => Some(Arc::new(previous_manifest)),
 977                     Err(err) => {
 978                         println!("Couldn't re-use previous manifest - {}", err);
 979                         None
 980                     }
 981                 }
 982             }
 983             Err(err) => {
 984                 println!("Couldn't download previous manifest - {}", err);
 985                 None
 986             }
 987         }
 988     } else {
 989         None
 990     };
 991
 992     let snapshot = BackupDir::new(backup_type, backup_id, backup_time)?;
 993     let mut manifest = BackupManifest::new(snapshot);
 994
 995     let mut catalog = None;
 996     let mut catalog_result_tx = None;
 997
 998     for (backup_type, filename, target, size) in upload_list {
 999         match backup_type {
1000             BackupSpecificationType::CONFIG => {
1001                 println!("Upload config file '{}' to '{}' as {}", filename, repo, target);
1002                 let stats = client
1003                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1004                     .await?;
1005                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1006             }
1007             BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
1008                 println!("Upload log file '{}' to '{}' as {}", filename, repo, target);
1009                 let stats = client
1010                     .upload_blob_from_file(&filename, &target, true, crypt_mode == CryptMode::Encrypt)
1011                     .await?;
1012                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1013             }
1014             BackupSpecificationType::PXAR => {
1015                 // start catalog upload on first use
1016                 if catalog.is_none() {
1017                     let (cat, res) = spawn_catalog_upload(client.clone(), crypt_mode == CryptMode::Encrypt)?;
1018                     catalog = Some(cat);
1019                     catalog_result_tx = Some(res);
1020                 }
1021                 let catalog = catalog.as_ref().unwrap();
1022
1023                 println!("Upload directory '{}' to '{}' as {}", filename, repo, target);
1024                 catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
1025                 let stats = backup_directory(
1026                     &client,
1027                     previous_manifest.clone(),
1028                     &filename,
1029                     &target,
1030                     chunk_size_opt,
1031                     devices.clone(),
1032                     verbose,
1033                     skip_lost_and_found,
1034                     catalog.clone(),
1035                     pattern_list.clone(),
1036                     entries_max as usize,
1037                     true,
1038                     crypt_mode == CryptMode::Encrypt,
1039                 ).await?;
1040                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1041                 catalog.lock().unwrap().end_directory()?;
1042             }
1043             BackupSpecificationType::IMAGE => {
1044                 println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
1045                 let stats = backup_image(
1046                     &client,
1047                     previous_manifest.clone(),
1048                      &filename,
1049                     &target,
1050                     size,
1051                     chunk_size_opt,
1052                     true,
1053                     crypt_mode == CryptMode::Encrypt,
1054                     verbose,
1055                 ).await?;
1056                 manifest.add_file(target, stats.size, stats.csum, crypt_mode)?;
1057             }
1058         }
1059     }
1060
1061     // finalize and upload catalog
1062     if let Some(catalog) = catalog {
1063         let mutex = Arc::try_unwrap(catalog)
1064             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
1065         let mut catalog = mutex.into_inner().unwrap();
1066
1067         catalog.finish()?;
1068
1069         drop(catalog); // close upload stream
1070
1071         if let Some(catalog_result_rx) = catalog_result_tx {
1072             let stats = catalog_result_rx.await??;
1073             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypt_mode)?;
1074         }
1075     }
1076
1077     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
1078         let target = ENCRYPTED_KEY_BLOB_NAME;
1079         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
1080         let stats = client
1081             .upload_blob_from_data(rsa_encrypted_key, target, false, false)
1082             .await?;
1083         manifest.add_file(target.to_string(), stats.size, stats.csum, crypt_mode)?;
1084
1085     }
1086     // create manifest (index.json)
1087     // manifests are never encrypted, but include a signature
1088     let manifest = manifest.to_string(crypt_config.as_ref().map(Arc::as_ref))
1089         .map_err(|err| format_err!("unable to format manifest - {}", err))?;
1090
1091
1092     if verbose { println!("Upload index.json to '{}'", repo) };
1093     client
1094         .upload_blob_from_data(manifest.into_bytes(), MANIFEST_BLOB_NAME, true, false)
1095         .await?;
1096
1097     client.finish().await?;
1098
1099     let end_time = std::time::Instant::now();
1100     let elapsed = end_time.duration_since(start_time);
1101     println!("Duration: {:.2}s", elapsed.as_secs_f64());
1102
1103     println!("End Time: {}", strftime_local("%c", epoch_i64())?);
1104
1105     Ok(Value::Null)
1106 }
1107
1108 fn complete_backup_source(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1109
1110     let mut result = vec![];
1111
1112     let data: Vec<&str> = arg.splitn(2, ':').collect();
1113
1114     if data.len() != 2 {
1115         result.push(String::from("root.pxar:/"));
1116         result.push(String::from("etc.pxar:/etc"));
1117         return result;
1118     }
1119
1120     let files = tools::complete_file_name(data[1], param);
1121
1122     for file in files {
1123         result.push(format!("{}:{}", data[0], file));
1124     }
1125
1126     result
1127 }
1128
1129 async fn dump_image<W: Write>(
1130     client: Arc<BackupReader>,
1131     crypt_config: Option<Arc<CryptConfig>>,
1132     crypt_mode: CryptMode,
1133     index: FixedIndexReader,
1134     mut writer: W,
1135     verbose: bool,
1136 ) -> Result<(), Error> {
1137
1138     let most_used = index.find_most_used_chunks(8);
1139
1140     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, crypt_mode, most_used);
1141
1142     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
1143     // and thus slows down reading. Instead, directly use RemoteChunkReader
1144     let mut per = 0;
1145     let mut bytes = 0;
1146     let start_time = std::time::Instant::now();
1147
1148     for pos in 0..index.index_count() {
1149         let digest = index.index_digest(pos).unwrap();
1150         let raw_data = chunk_reader.read_chunk(&digest).await?;
1151         writer.write_all(&raw_data)?;
1152         bytes += raw_data.len();
1153         if verbose {
1154             let next_per = ((pos+1)*100)/index.index_count();
1155             if per != next_per {
1156                 eprintln!("progress {}% (read {} bytes, duration {} sec)",
1157                           next_per, bytes, start_time.elapsed().as_secs());
1158                 per = next_per;
1159             }
1160         }
1161     }
1162
1163     let end_time = std::time::Instant::now();
1164     let elapsed = end_time.duration_since(start_time);
1165     eprintln!("restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1166               bytes,
1167               elapsed.as_secs_f64(),
1168               bytes as f64/(1024.0*1024.0*elapsed.as_secs_f64())
1169     );
1170
1171
1172     Ok(())
1173 }
1174
1175 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1176     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1177         (name.into(), archive_type(name).unwrap())
1178     } else if name.ends_with(".pxar") {
1179         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1180     } else if name.ends_with(".img") {
1181         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1182     } else {
1183         (format!("{}.blob", name), ArchiveType::Blob)
1184     }
1185 }
1186
1187 #[api(
1188    input: {
1189        properties: {
1190            repository: {
1191                schema: REPO_URL_SCHEMA,
1192                optional: true,
1193            },
1194            snapshot: {
1195                type: String,
1196                description: "Group/Snapshot path.",
1197            },
1198            "archive-name": {
1199                description: "Backup archive name.",
1200                type: String,
1201            },
1202            target: {
1203                type: String,
1204                description: r###"Target directory path. Use '-' to write to standard output.
1205
1206 We do not extraxt '.pxar' archives when writing to standard output.
1207
1208 "###
1209            },
1210            "allow-existing-dirs": {
1211                type: Boolean,
1212                description: "Do not fail if directories already exists.",
1213                optional: true,
1214            },
1215            keyfile: {
1216                schema: KEYFILE_SCHEMA,
1217                optional: true,
1218            },
1219            "keyfd": {
1220                schema: KEYFD_SCHEMA,
1221                optional: true,
1222            },
1223            "crypt-mode": {
1224                type: CryptMode,
1225                optional: true,
1226            },
1227        }
1228    }
1229 )]
1230 /// Restore backup repository.
1231 async fn restore(param: Value) -> Result<Value, Error> {
1232     let repo = extract_repository_from_value(&param)?;
1233
1234     let verbose = param["verbose"].as_bool().unwrap_or(false);
1235
1236     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1237
1238     let archive_name = tools::required_string_param(&param, "archive-name")?;
1239
1240     let client = connect(&repo)?;
1241
1242     record_repository(&repo);
1243
1244     let path = tools::required_string_param(&param, "snapshot")?;
1245
1246     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
1247         let group: BackupGroup = path.parse()?;
1248         api_datastore_latest_snapshot(&client, repo.store(), group).await?
1249     } else {
1250         let snapshot: BackupDir = path.parse()?;
1251         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
1252     };
1253
1254     let target = tools::required_string_param(&param, "target")?;
1255     let target = if target == "-" { None } else { Some(target) };
1256
1257     let (keydata, _crypt_mode) = keyfile_parameters(&param)?;
1258
1259     let crypt_config = match keydata {
1260         None => None,
1261         Some(key) => {
1262             let (key, _, fingerprint) = decrypt_key(&key, &key::get_encryption_key_password)?;
1263             eprintln!("Encryption key fingerprint: '{}'", fingerprint);
1264             Some(Arc::new(CryptConfig::new(key)?))
1265         }
1266     };
1267
1268     let client = BackupReader::start(
1269         client,
1270         crypt_config.clone(),
1271         repo.store(),
1272         &backup_type,
1273         &backup_id,
1274         backup_time,
1275         true,
1276     ).await?;
1277
1278     let (archive_name, archive_type) = parse_archive_type(archive_name);
1279
1280     let (manifest, backup_index_data) = client.download_manifest().await?;
1281
1282     if archive_name == ENCRYPTED_KEY_BLOB_NAME && crypt_config.is_none() {
1283         eprintln!("Restoring encrypted key blob without original key - skipping manifest fingerprint check!")
1284     } else {
1285         manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref))?;
1286     }
1287
1288     if archive_name == MANIFEST_BLOB_NAME {
1289         if let Some(target) = target {
1290             replace_file(target, &backup_index_data, CreateOptions::new())?;
1291         } else {
1292             let stdout = std::io::stdout();
1293             let mut writer = stdout.lock();
1294             writer.write_all(&backup_index_data)
1295                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1296         }
1297
1298         return Ok(Value::Null);
1299     }
1300
1301     let file_info = manifest.lookup_file_info(&archive_name)?;
1302
1303     if archive_type == ArchiveType::Blob {
1304
1305         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1306
1307         if let Some(target) = target {
1308            let mut writer = std::fs::OpenOptions::new()
1309                 .write(true)
1310                 .create(true)
1311                 .create_new(true)
1312                 .open(target)
1313                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?;
1314             std::io::copy(&mut reader, &mut writer)?;
1315         } else {
1316             let stdout = std::io::stdout();
1317             let mut writer = stdout.lock();
1318             std::io::copy(&mut reader, &mut writer)
1319                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1320         }
1321
1322     } else if archive_type == ArchiveType::DynamicIndex {
1323
1324         let index = client.download_dynamic_index(&manifest, &archive_name).await?;
1325
1326         let most_used = index.find_most_used_chunks(8);
1327
1328         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, file_info.chunk_crypt_mode(), most_used);
1329
1330         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1331
1332         if let Some(target) = target {
1333             proxmox_backup::pxar::extract_archive(
1334                 pxar::decoder::Decoder::from_std(reader)?,
1335                 Path::new(target),
1336                 &[],
1337                 true,
1338                 proxmox_backup::pxar::Flags::DEFAULT,
1339                 allow_existing_dirs,
1340                 |path| {
1341                     if verbose {
1342                         println!("{:?}", path);
1343                     }
1344                 },
1345                 None,
1346             )
1347             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1348         } else {
1349             let mut writer = std::fs::OpenOptions::new()
1350                 .write(true)
1351                 .open("/dev/stdout")
1352                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1353
1354             std::io::copy(&mut reader, &mut writer)
1355                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1356         }
1357     } else if archive_type == ArchiveType::FixedIndex {
1358
1359         let index = client.download_fixed_index(&manifest, &archive_name).await?;
1360
1361         let mut writer = if let Some(target) = target {
1362             std::fs::OpenOptions::new()
1363                 .write(true)
1364                 .create(true)
1365                 .create_new(true)
1366                 .open(target)
1367                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1368         } else {
1369             std::fs::OpenOptions::new()
1370                 .write(true)
1371                 .open("/dev/stdout")
1372                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1373         };
1374
1375         dump_image(client.clone(), crypt_config.clone(), file_info.chunk_crypt_mode(), index, &mut writer, verbose).await?;
1376     }
1377
1378     Ok(Value::Null)
1379 }
1380
1381 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
1382     &ApiHandler::Async(&prune),
1383     &ObjectSchema::new(
1384         "Prune a backup repository.",
1385         &proxmox_backup::add_common_prune_prameters!([
1386             ("dry-run", true, &BooleanSchema::new(
1387                 "Just show what prune would do, but do not delete anything.")
1388              .schema()),
1389             ("group", false, &StringSchema::new("Backup group.").schema()),
1390         ], [
1391             ("output-format", true, &OUTPUT_FORMAT),
1392             (
1393                 "quiet",
1394                 true,
1395                 &BooleanSchema::new("Minimal output - only show removals.")
1396                     .schema()
1397             ),
1398             ("repository", true, &REPO_URL_SCHEMA),
1399         ])
1400     )
1401 );
1402
1403 fn prune<'a>(
1404     param: Value,
1405     _info: &ApiMethod,
1406     _rpcenv: &'a mut dyn RpcEnvironment,
1407 ) -> proxmox::api::ApiFuture<'a> {
1408     async move {
1409         prune_async(param).await
1410     }.boxed()
1411 }
1412
1413 async fn prune_async(mut param: Value) -> Result<Value, Error> {
1414     let repo = extract_repository_from_value(&param)?;
1415
1416     let mut client = connect(&repo)?;
1417
1418     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1419
1420     let group = tools::required_string_param(&param, "group")?;
1421     let group: BackupGroup = group.parse()?;
1422
1423     let output_format = get_output_format(&param);
1424
1425     let quiet = param["quiet"].as_bool().unwrap_or(false);
1426
1427     param.as_object_mut().unwrap().remove("repository");
1428     param.as_object_mut().unwrap().remove("group");
1429     param.as_object_mut().unwrap().remove("output-format");
1430     param.as_object_mut().unwrap().remove("quiet");
1431
1432     param["backup-type"] = group.backup_type().into();
1433     param["backup-id"] = group.backup_id().into();
1434
1435     let mut result = client.post(&path, Some(param)).await?;
1436
1437     record_repository(&repo);
1438
1439     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1440         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1441         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time)?;
1442         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
1443     };
1444
1445     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1446         Ok(match v.as_bool() {
1447             Some(true) => "keep",
1448             Some(false) => "remove",
1449             None => "unknown",
1450         }.to_string())
1451     };
1452
1453     let options = default_table_format_options()
1454         .sortby("backup-type", false)
1455         .sortby("backup-id", false)
1456         .sortby("backup-time", false)
1457         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
1458         .column(ColumnConfig::new("backup-time").renderer(tools::format::render_epoch).header("date"))
1459         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
1460         ;
1461
1462     let return_type = &proxmox_backup::api2::admin::datastore::API_METHOD_PRUNE.returns;
1463
1464     let mut data = result["data"].take();
1465
1466     if quiet {
1467         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
1468             item["keep"].as_bool() == Some(false)
1469         }).map(|v| v.clone()).collect();
1470         data = list.into();
1471     }
1472
1473     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1474
1475     Ok(Value::Null)
1476 }
1477
1478 #[api(
1479    input: {
1480        properties: {
1481            repository: {
1482                schema: REPO_URL_SCHEMA,
1483                optional: true,
1484            },
1485            "output-format": {
1486                schema: OUTPUT_FORMAT,
1487                optional: true,
1488            },
1489        }
1490    },
1491     returns: {
1492         type: StorageStatus,
1493     },
1494 )]
1495 /// Get repository status.
1496 async fn status(param: Value) -> Result<Value, Error> {
1497
1498     let repo = extract_repository_from_value(&param)?;
1499
1500     let output_format = get_output_format(&param);
1501
1502     let client = connect(&repo)?;
1503
1504     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1505
1506     let mut result = client.get(&path, None).await?;
1507     let mut data = result["data"].take();
1508
1509     record_repository(&repo);
1510
1511     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1512         let v = v.as_u64().unwrap();
1513         let total = record["total"].as_u64().unwrap();
1514         let roundup = total/200;
1515         let per = ((v+roundup)*100)/total;
1516         let info = format!(" ({} %)", per);
1517         Ok(format!("{} {:>8}", v, info))
1518     };
1519
1520     let options = default_table_format_options()
1521         .noheader(true)
1522         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1523         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1524         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1525
1526     let return_type = &API_METHOD_STATUS.returns;
1527
1528     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1529
1530     Ok(Value::Null)
1531 }
1532
1533 // like get, but simply ignore errors and return Null instead
1534 async fn try_get(repo: &BackupRepository, url: &str) -> Value {
1535
1536     let fingerprint = std::env::var(ENV_VAR_PBS_FINGERPRINT).ok();
1537     let password = std::env::var(ENV_VAR_PBS_PASSWORD).ok();
1538
1539     let options = HttpClientOptions::new()
1540         .prefix(Some("proxmox-backup".to_string()))
1541         .password(password)
1542         .interactive(false)
1543         .fingerprint(fingerprint)
1544         .fingerprint_cache(true)
1545         .ticket_cache(true);
1546
1547     let client = match HttpClient::new(repo.host(), repo.port(), repo.auth_id(), options) {
1548         Ok(v) => v,
1549         _ => return Value::Null,
1550     };
1551
1552     let mut resp = match client.get(url, None).await {
1553         Ok(v) => v,
1554         _ => return Value::Null,
1555     };
1556
1557     if let Some(map) = resp.as_object_mut() {
1558         if let Some(data) = map.remove("data") {
1559             return data;
1560         }
1561     }
1562     Value::Null
1563 }
1564
1565 fn complete_backup_group(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1566     proxmox_backup::tools::runtime::main(async { complete_backup_group_do(param).await })
1567 }
1568
1569 async fn complete_backup_group_do(param: &HashMap<String, String>) -> Vec<String> {
1570
1571     let mut result = vec![];
1572
1573     let repo = match extract_repository_from_map(param) {
1574         Some(v) => v,
1575         _ => return result,
1576     };
1577
1578     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
1579
1580     let data = try_get(&repo, &path).await;
1581
1582     if let Some(list) = data.as_array() {
1583         for item in list {
1584             if let (Some(backup_id), Some(backup_type)) =
1585                 (item["backup-id"].as_str(), item["backup-type"].as_str())
1586             {
1587                 result.push(format!("{}/{}", backup_type, backup_id));
1588             }
1589         }
1590     }
1591
1592     result
1593 }
1594
1595 pub fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1596     proxmox_backup::tools::runtime::main(async { complete_group_or_snapshot_do(arg, param).await })
1597 }
1598
1599 async fn complete_group_or_snapshot_do(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1600
1601     if arg.matches('/').count() < 2 {
1602         let groups = complete_backup_group_do(param).await;
1603         let mut result = vec![];
1604         for group in groups {
1605             result.push(group.to_string());
1606             result.push(format!("{}/", group));
1607         }
1608         return result;
1609     }
1610
1611     complete_backup_snapshot_do(param).await
1612 }
1613
1614 fn complete_backup_snapshot(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1615     proxmox_backup::tools::runtime::main(async { complete_backup_snapshot_do(param).await })
1616 }
1617
1618 async fn complete_backup_snapshot_do(param: &HashMap<String, String>) -> Vec<String> {
1619
1620     let mut result = vec![];
1621
1622     let repo = match extract_repository_from_map(param) {
1623         Some(v) => v,
1624         _ => return result,
1625     };
1626
1627     let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
1628
1629     let data = try_get(&repo, &path).await;
1630
1631     if let Some(list) = data.as_array() {
1632         for item in list {
1633             if let (Some(backup_id), Some(backup_type), Some(backup_time)) =
1634                 (item["backup-id"].as_str(), item["backup-type"].as_str(), item["backup-time"].as_i64())
1635             {
1636                 if let Ok(snapshot) = BackupDir::new(backup_type, backup_id, backup_time) {
1637                     result.push(snapshot.relative_path().to_str().unwrap().to_owned());
1638                 }
1639             }
1640         }
1641     }
1642
1643     result
1644 }
1645
1646 fn complete_server_file_name(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1647     proxmox_backup::tools::runtime::main(async { complete_server_file_name_do(param).await })
1648 }
1649
1650 async fn complete_server_file_name_do(param: &HashMap<String, String>) -> Vec<String> {
1651
1652     let mut result = vec![];
1653
1654     let repo = match extract_repository_from_map(param) {
1655         Some(v) => v,
1656         _ => return result,
1657     };
1658
1659     let snapshot: BackupDir = match param.get("snapshot") {
1660         Some(path) => {
1661             match path.parse() {
1662                 Ok(v) => v,
1663                 _ => return result,
1664             }
1665         }
1666         _ => return result,
1667     };
1668
1669     let query = tools::json_object_to_query(json!({
1670         "backup-type": snapshot.group().backup_type(),
1671         "backup-id": snapshot.group().backup_id(),
1672         "backup-time": snapshot.backup_time(),
1673     })).unwrap();
1674
1675     let path = format!("api2/json/admin/datastore/{}/files?{}", repo.store(), query);
1676
1677     let data = try_get(&repo, &path).await;
1678
1679     if let Some(list) = data.as_array() {
1680         for item in list {
1681             if let Some(filename) = item["filename"].as_str() {
1682                 result.push(filename.to_owned());
1683             }
1684         }
1685     }
1686
1687     result
1688 }
1689
1690 fn complete_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1691     complete_server_file_name(arg, param)
1692         .iter()
1693         .map(|v| tools::format::strip_server_file_extension(&v))
1694         .collect()
1695 }
1696
1697 pub fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1698     complete_server_file_name(arg, param)
1699         .iter()
1700         .filter_map(|name| {
1701             if name.ends_with(".pxar.didx") {
1702                 Some(tools::format::strip_server_file_extension(name))
1703             } else {
1704                 None
1705             }
1706         })
1707         .collect()
1708 }
1709
1710 pub fn complete_img_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1711     complete_server_file_name(arg, param)
1712         .iter()
1713         .filter_map(|name| {
1714             if name.ends_with(".img.fidx") {
1715                 Some(tools::format::strip_server_file_extension(name))
1716             } else {
1717                 None
1718             }
1719         })
1720         .collect()
1721 }
1722
1723 fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
1724
1725     let mut result = vec![];
1726
1727     let mut size = 64;
1728     loop {
1729         result.push(size.to_string());
1730         size *= 2;
1731         if size > 4096 { break; }
1732     }
1733
1734     result
1735 }
1736
1737 fn complete_auth_id(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
1738     proxmox_backup::tools::runtime::main(async { complete_auth_id_do(param).await })
1739 }
1740
1741 async fn complete_auth_id_do(param: &HashMap<String, String>) -> Vec<String> {
1742
1743     let mut result = vec![];
1744
1745     let repo = match extract_repository_from_map(param) {
1746         Some(v) => v,
1747         _ => return result,
1748     };
1749
1750     let data = try_get(&repo, "api2/json/access/users?include_tokens=true").await;
1751
1752     if let Ok(parsed) = serde_json::from_value::<Vec<UserWithTokens>>(data) {
1753         for user in parsed {
1754             result.push(user.userid.to_string());
1755             for token in user.tokens {
1756                 result.push(token.tokenid.to_string());
1757             }
1758         }
1759     };
1760
1761     result
1762 }
1763
1764 use proxmox_backup::client::RemoteChunkReader;
1765 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
1766 /// async use!
1767 ///
1768 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
1769 /// so that we can properly access it from multiple threads simultaneously while not issuing
1770 /// duplicate simultaneous reads over http.
1771 pub struct BufferedDynamicReadAt {
1772     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
1773 }
1774
1775 impl BufferedDynamicReadAt {
1776     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
1777         Self {
1778             inner: Mutex::new(inner),
1779         }
1780     }
1781 }
1782
1783 impl ReadAt for BufferedDynamicReadAt {
1784     fn start_read_at<'a>(
1785         self: Pin<&'a Self>,
1786         _cx: &mut Context,
1787         buf: &'a mut [u8],
1788         offset: u64,
1789     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1790         MaybeReady::Ready(tokio::task::block_in_place(move || {
1791             let mut reader = self.inner.lock().unwrap();
1792             reader.seek(SeekFrom::Start(offset))?;
1793             Ok(reader.read(buf)?)
1794         }))
1795     }
1796
1797     fn poll_complete<'a>(
1798         self: Pin<&'a Self>,
1799         _op: ReadAtOperation<'a>,
1800     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1801         panic!("LocalDynamicReadAt::start_read_at returned Pending");
1802     }
1803 }
1804
1805 fn main() {
1806
1807     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
1808         .arg_param(&["backupspec"])
1809         .completion_cb("repository", complete_repository)
1810         .completion_cb("backupspec", complete_backup_source)
1811         .completion_cb("keyfile", tools::complete_file_name)
1812         .completion_cb("chunk-size", complete_chunk_size);
1813
1814     let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
1815         .completion_cb("repository", complete_repository)
1816         .completion_cb("keyfile", tools::complete_file_name);
1817
1818     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
1819         .completion_cb("repository", complete_repository);
1820
1821     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
1822         .completion_cb("repository", complete_repository);
1823
1824     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
1825         .arg_param(&["snapshot", "archive-name", "target"])
1826         .completion_cb("repository", complete_repository)
1827         .completion_cb("snapshot", complete_group_or_snapshot)
1828         .completion_cb("archive-name", complete_archive_name)
1829         .completion_cb("target", tools::complete_file_name);
1830
1831     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
1832         .arg_param(&["group"])
1833         .completion_cb("group", complete_backup_group)
1834         .completion_cb("repository", complete_repository);
1835
1836     let status_cmd_def = CliCommand::new(&API_METHOD_STATUS)
1837         .completion_cb("repository", complete_repository);
1838
1839     let login_cmd_def = CliCommand::new(&API_METHOD_API_LOGIN)
1840         .completion_cb("repository", complete_repository);
1841
1842     let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
1843         .completion_cb("repository", complete_repository);
1844
1845     let version_cmd_def = CliCommand::new(&API_METHOD_API_VERSION)
1846         .completion_cb("repository", complete_repository);
1847
1848     let change_owner_cmd_def = CliCommand::new(&API_METHOD_CHANGE_BACKUP_OWNER)
1849         .arg_param(&["group", "new-owner"])
1850         .completion_cb("group", complete_backup_group)
1851         .completion_cb("new-owner",  complete_auth_id)
1852         .completion_cb("repository", complete_repository);
1853
1854     let cmd_def = CliCommandMap::new()
1855         .insert("backup", backup_cmd_def)
1856         .insert("garbage-collect", garbage_collect_cmd_def)
1857         .insert("list", list_cmd_def)
1858         .insert("login", login_cmd_def)
1859         .insert("logout", logout_cmd_def)
1860         .insert("prune", prune_cmd_def)
1861         .insert("restore", restore_cmd_def)
1862         .insert("snapshot", snapshot_mgtm_cli())
1863         .insert("status", status_cmd_def)
1864         .insert("key", key::cli())
1865         .insert("mount", mount_cmd_def())
1866         .insert("map", map_cmd_def())
1867         .insert("unmap", unmap_cmd_def())
1868         .insert("catalog", catalog_mgmt_cli())
1869         .insert("task", task_mgmt_cli())
1870         .insert("version", version_cmd_def)
1871         .insert("benchmark", benchmark_cmd_def)
1872         .insert("change-owner", change_owner_cmd_def)
1873
1874         .alias(&["files"], &["snapshot", "files"])
1875         .alias(&["forget"], &["snapshot", "forget"])
1876         .alias(&["upload-log"], &["snapshot", "upload-log"])
1877         .alias(&["snapshots"], &["snapshot", "list"])
1878         ;
1879
1880     let rpcenv = CliEnvironment::new();
1881     run_cli_command(cmd_def, rpcenv, Some(|future| {
1882         proxmox_backup::tools::runtime::main(future)
1883     }));
1884 }