s3m-core 0.18.0

Reusable S3 client and streaming engine behind the s3m CLI
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

use crate::{
    s3::RequestOptions,
    s3::S3,
    stream::{
        InitialStreamParams, STDIN_BUFFER_SIZE, Stream, complete_multipart_upload,
        create_initial_stream, create_nonce_header, encrypt_chunk, get_key, init_encryption,
        initiate_multipart_upload, maybe_upload_part, setup_stream_progress, upload_final_part,
        write_to_stream,
    },
};
use anyhow::{Result, anyhow};
use chacha20poly1305::aead::stream::EncryptorBE32;
use futures::stream::TryStreamExt;
use std::{collections::BTreeMap, path::PathBuf};
use tokio::io::stdin;
use tokio_util::codec::{BytesCodec, FramedRead};

/// Read from STDIN and encrypt the data.
///
/// # Errors
/// Will return an error if the upload fails.
pub async fn stream_stdin_encrypted(
    s3: &S3,
    object_key: &str,
    acl: Option<String>,
    meta: Option<BTreeMap<String, String>>,
    quiet: bool,
    tmp_dir: PathBuf,
    globals: RequestOptions,
) -> Result<String> {
    let encryption_key = globals
        .enc_key
        .as_ref()
        .ok_or_else(|| anyhow!("Encryption key is required"))?;

    let key = get_key(object_key, globals.compress, globals.encrypt);

    let mut meta = meta.unwrap_or_default();
    meta.insert(
        "Content-Type".to_string(),
        "application/vnd.s3m.encrypted".to_string(),
    );

    let upload_id = initiate_multipart_upload(s3, &key, acl, meta).await?;
    let progress_sender = setup_stream_progress(quiet).await;

    let (cipher, nonce_bytes) = init_encryption(encryption_key);
    let encryptor = EncryptorBE32::from_aead(cipher, (&nonce_bytes).into());
    let nonce_header = create_nonce_header(&nonce_bytes);

    let first_stream: Stream = create_initial_stream(InitialStreamParams {
        upload_id: &upload_id,
        tmp_dir: &tmp_dir,
        key: &key,
        s3,
        progress_sender,
        globals: &globals,
        header_data: Some(&nonce_header),
    })?;

    let mut stream = FramedRead::new(stdin(), BytesCodec::new())
        .map_err(|e| anyhow!("Error reading STDIN chunk: {e}"))
        .try_fold(
            (first_stream, encryptor),
            |(mut current_upload_state_acc, mut current_encryptor_acc), chunk| async move {
                let encrypted_data = encrypt_chunk(&mut current_encryptor_acc, &chunk)
                    .map_err(|e| anyhow!("Failed to encrypt chunk: {e}"))?;

                write_to_stream(&mut current_upload_state_acc, &encrypted_data).map_err(|e| {
                    anyhow!("Failed to write encrypted chunk to upload stream: {e}")
                })?;

                maybe_upload_part(&mut current_upload_state_acc, STDIN_BUFFER_SIZE).await?;

                Ok((current_upload_state_acc, current_encryptor_acc))
            },
        )
        .await
        .map(|(final_stream_state, _)| final_stream_state)?;

    let final_etag = upload_final_part(&mut stream, &key, &upload_id, s3, &globals).await?;
    stream.etags.push(final_etag);

    if let Some(sender) = stream.channel.take() {
        drop(sender);
    }

    complete_multipart_upload(s3, &key, &upload_id, stream.etags).await
}