src/bin/proxmox-backup-client.rs

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