use std::io::Write as _;
use std::path::Path;
use anyhow::{Context, Result};
use aws_sdk_s3::operation::get_object_annotation::GetObjectAnnotationOutput;
use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumType, ServerSideEncryption};
use tempfile::NamedTempFile;
use tracing::{info, warn};
use s3util_rs::config::ClientConfig;
use s3util_rs::config::args::get_object_annotation::GetObjectAnnotationArgs;
use s3util_rs::output::json::get_object_annotation_to_json;
use s3util_rs::storage::annotation;
use s3util_rs::storage::checksum::AdditionalChecksum;
use s3util_rs::storage::s3::api::{self, GetObjectAnnotationParams, ObjectAnnotationError};
use super::ExitStatus;
fn detect_checksum(out: &GetObjectAnnotationOutput) -> Option<(ChecksumAlgorithm, String)> {
if let Some(v) = out.checksum_crc64_nvme() {
return Some((ChecksumAlgorithm::Crc64Nvme, v.to_string()));
}
if let Some(v) = out.checksum_crc32() {
return Some((ChecksumAlgorithm::Crc32, v.to_string()));
}
if let Some(v) = out.checksum_crc32_c() {
return Some((ChecksumAlgorithm::Crc32C, v.to_string()));
}
if let Some(v) = out.checksum_sha1() {
return Some((ChecksumAlgorithm::Sha1, v.to_string()));
}
if let Some(v) = out.checksum_sha256() {
return Some((ChecksumAlgorithm::Sha256, v.to_string()));
}
if let Some(v) = out.checksum_sha512() {
return Some((ChecksumAlgorithm::Sha512, v.to_string()));
}
if let Some(v) = out.checksum_md5() {
return Some((ChecksumAlgorithm::Md5, v.to_string()));
}
if let Some(v) = out.checksum_xxhash64() {
return Some((ChecksumAlgorithm::Xxhash64, v.to_string()));
}
if let Some(v) = out.checksum_xxhash3() {
return Some((ChecksumAlgorithm::Xxhash3, v.to_string()));
}
if let Some(v) = out.checksum_xxhash128() {
return Some((ChecksumAlgorithm::Xxhash128, v.to_string()));
}
None
}
#[derive(Debug)]
enum IntegrityCheck {
Verified,
Unverifiable,
}
fn check_integrity(
bytes: &[u8],
content_length: Option<i64>,
e_tag: Option<&str>,
sse: Option<&ServerSideEncryption>,
checksum: Option<&(ChecksumAlgorithm, String)>,
bucket: &str,
key: &str,
) -> Result<IntegrityCheck> {
if let Some(len) = content_length
&& len != bytes.len() as i64
{
anyhow::bail!(
"content length mismatch for s3://{bucket}/{key}: response said {len} bytes, received {}",
bytes.len()
);
}
let mut verified = false;
match annotation::verify_etag_md5(bytes, e_tag, sse) {
Some(true) => verified = true,
Some(false) => anyhow::bail!("ETag (MD5) verification failed for s3://{bucket}/{key}"),
None => {}
}
if let Some((algo, expected)) = checksum {
if !AdditionalChecksum::is_supported(algo) {
anyhow::bail!(
"s3://{bucket}/{key} carries a {} checksum, which s3util cannot verify",
algo.as_str()
);
}
if annotation::verify_additional_checksum(bytes, algo.clone(), expected) {
verified = true;
} else {
anyhow::bail!(
"{} checksum verification failed for s3://{bucket}/{key}",
algo.as_str()
);
}
}
Ok(if verified {
IntegrityCheck::Verified
} else {
IntegrityCheck::Unverifiable
})
}
fn verify_saved_file(
path: &Path,
content_length: Option<i64>,
e_tag: Option<&str>,
sse: Option<&ServerSideEncryption>,
checksum: Option<&(ChecksumAlgorithm, String)>,
bucket: &str,
key: &str,
) -> Result<()> {
let on_disk = std::fs::read(path).with_context(|| {
format!(
"re-reading the freshly-written temp file {} for verification",
path.display()
)
})?;
check_integrity(&on_disk, content_length, e_tag, sse, checksum, bucket, key).with_context(
|| {
format!(
"post-write verification failed for s3://{bucket}/{key}: the payload was corrupted while writing to disk"
)
},
)?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn write_verified_output(
outfile: &str,
payload: &[u8],
content_length: Option<i64>,
e_tag: Option<&str>,
sse: Option<&ServerSideEncryption>,
checksum: Option<&(ChecksumAlgorithm, String)>,
bucket: &str,
key: &str,
) -> Result<()> {
let path = Path::new(outfile);
let parent = match path.parent() {
Some(p) if !p.as_os_str().is_empty() => p,
_ => Path::new("."),
};
let mut tmp = NamedTempFile::new_in(parent)
.with_context(|| format!("creating temp file next to {outfile}"))?;
tmp.write_all(payload)
.context("writing annotation payload to temp file")?;
tmp.flush()
.context("flushing annotation payload temp file")?;
verify_saved_file(
tmp.path(),
content_length,
e_tag,
sse,
checksum,
bucket,
key,
)
.with_context(|| format!("aborted saving to {outfile}; the destination was left untouched"))?;
tmp.persist(path)
.map_err(|e| anyhow::anyhow!("persisting annotation payload to {outfile}: {e}"))?;
Ok(())
}
pub async fn run_get_object_annotation(
args: GetObjectAnnotationArgs,
client_config: ClientConfig,
) -> Result<ExitStatus> {
let (bucket, key) = args
.bucket_key()
.map_err(|e| anyhow::anyhow!("{}", e.trim_end()))?;
let annotation_name = args
.annotation_name
.as_deref()
.ok_or_else(|| anyhow::anyhow!("--annotation-name is required"))?;
annotation::validate_annotation_name(annotation_name)?;
let outfile = args
.outfile
.as_deref()
.ok_or_else(|| anyhow::anyhow!("outfile is required"))?;
let request_payer = client_config.request_payer.clone();
let client = client_config.create_client().await;
let params = GetObjectAnnotationParams {
bucket: &bucket,
key: &key,
annotation_name,
version_id: args.target_version_id.as_deref(),
request_payer,
};
let out = match api::get_object_annotation(&client, params).await {
Ok(out) => out,
Err(ObjectAnnotationError::BucketNotFound) => {
tracing::error!("bucket s3://{bucket} not found");
return Ok(ExitStatus::NotFound);
}
Err(ObjectAnnotationError::NotFound) => {
match args.target_version_id.as_deref() {
Some(v) => tracing::error!("s3://{bucket}/{key} (versionId={v}) not found"),
None => tracing::error!("object s3://{bucket}/{key} not found"),
}
return Ok(ExitStatus::NotFound);
}
Err(ObjectAnnotationError::AnnotationNotFound) => {
match args.target_version_id.as_deref() {
Some(v) => tracing::error!(
"annotation {annotation_name} not found for s3://{bucket}/{key} (versionId={v})"
),
None => {
tracing::error!(
"annotation {annotation_name} not found for s3://{bucket}/{key}"
)
}
}
return Ok(ExitStatus::NotFound);
}
Err(ObjectAnnotationError::Other(e)) => return Err(e),
};
let json = get_object_annotation_to_json(&out);
let e_tag = out.e_tag().map(str::to_string);
let sse = out.server_side_encryption().cloned();
let content_length = out.content_length();
let is_composite = matches!(out.checksum_type(), Some(ct) if *ct == ChecksumType::Composite);
let checksum = if is_composite {
None
} else {
detect_checksum(&out)
};
let mut body = out.annotation_payload;
let cap = annotation::MAX_ANNOTATION_PAYLOAD_LEN;
let mut payload: Vec<u8> = Vec::new();
while let Some(chunk) = body
.try_next()
.await
.context("reading annotation payload body")?
{
payload.extend_from_slice(&chunk);
if payload.len() > cap {
anyhow::bail!(
"annotation payload for s3://{bucket}/{key} exceeds the 1 MiB limit ({cap} bytes)"
);
}
}
let verified = match check_integrity(
&payload,
content_length,
e_tag.as_deref(),
sse.as_ref(),
checksum.as_ref(),
&bucket,
&key,
)? {
IntegrityCheck::Verified => true,
IntegrityCheck::Unverifiable => {
warn!(
bucket = %bucket,
key = %key,
"payload integrity could not be verified (no AES256 ETag and no additional checksum)."
);
false
}
};
if outfile == "-" {
std::io::stdout()
.write_all(&payload)
.context("writing annotation payload to stdout")?;
return Ok(ExitStatus::Success);
}
write_verified_output(
outfile,
&payload,
content_length,
e_tag.as_deref(),
sse.as_ref(),
checksum.as_ref(),
&bucket,
&key,
)?;
println!("{}", serde_json::to_string_pretty(&json)?);
let outcome = if verified {
"written and verified"
} else {
"written, but integrity could NOT be verified"
};
info!(
bucket = %bucket,
key = %key,
annotation_name = %annotation_name,
outfile = %outfile,
"Annotation payload {}.",
outcome
);
Ok(ExitStatus::Success)
}
#[cfg(test)]
mod tests {
use super::*;
use aws_sdk_s3::primitives::ByteStream;
fn empty_payload() -> ByteStream {
ByteStream::from_static(b"")
}
#[test]
fn detect_checksum_none_when_no_checksum_present() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.build();
assert!(detect_checksum(&out).is_none());
}
#[test]
fn detect_checksum_crc64nvme() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_crc64_nvme("crc64val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Crc64Nvme);
assert_eq!(val, "crc64val");
}
#[test]
fn detect_checksum_crc32() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_crc32("crc32val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Crc32);
assert_eq!(val, "crc32val");
}
#[test]
fn detect_checksum_crc32c() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_crc32_c("crc32cval")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Crc32C);
assert_eq!(val, "crc32cval");
}
#[test]
fn detect_checksum_sha1() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_sha1("sha1val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Sha1);
assert_eq!(val, "sha1val");
}
#[test]
fn detect_checksum_sha256() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_sha256("sha256val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Sha256);
assert_eq!(val, "sha256val");
}
#[test]
fn detect_checksum_prefers_crc64nvme_over_others() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_crc64_nvme("crc64val")
.checksum_crc32("crc32val")
.checksum_crc32_c("crc32cval")
.checksum_sha1("sha1val")
.checksum_sha256("sha256val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Crc64Nvme);
assert_eq!(val, "crc64val");
}
#[test]
fn detect_checksum_prefers_crc32_over_sha() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_crc32("crc32val")
.checksum_sha1("sha1val")
.checksum_sha256("sha256val")
.build();
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Crc32);
assert_eq!(val, "crc32val");
}
#[test]
fn detect_checksum_surfaces_every_unsupported_algorithm() {
let cases = [
(
GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_sha512("v")
.build(),
ChecksumAlgorithm::Sha512,
),
(
GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_md5("v")
.build(),
ChecksumAlgorithm::Md5,
),
(
GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_xxhash64("v")
.build(),
ChecksumAlgorithm::Xxhash64,
),
(
GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_xxhash3("v")
.build(),
ChecksumAlgorithm::Xxhash3,
),
(
GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_xxhash128("v")
.build(),
ChecksumAlgorithm::Xxhash128,
),
];
for (out, expected) in cases {
let (algo, val) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, expected, "unexpected algorithm for {expected:?}");
assert_eq!(val, "v");
}
}
#[test]
fn detect_checksum_prefers_supported_over_unsupported() {
let out = GetObjectAnnotationOutput::builder()
.annotation_payload(empty_payload())
.checksum_sha256("sha256val")
.checksum_sha512("sha512val")
.build();
let (algo, _) = detect_checksum(&out).expect("checksum present");
assert_eq!(algo, ChecksumAlgorithm::Sha256);
}
#[test]
fn check_integrity_verified_on_matching_checksum() {
let payload = b"hello world";
let checksum = (
ChecksumAlgorithm::Crc64Nvme,
annotation::compute_checksum_base64(payload, ChecksumAlgorithm::Crc64Nvme),
);
let res = check_integrity(
payload,
Some(payload.len() as i64),
None,
None,
Some(&checksum),
"b",
"k",
)
.unwrap();
assert!(matches!(res, IntegrityCheck::Verified));
}
#[test]
fn check_integrity_verified_on_matching_aes256_etag() {
let payload = b"hello";
let etag = format!("\"{:x}\"", md5::compute(payload));
let res = check_integrity(
payload,
None,
Some(&etag),
Some(&ServerSideEncryption::Aes256),
None,
"b",
"k",
)
.unwrap();
assert!(matches!(res, IntegrityCheck::Verified));
}
#[test]
fn check_integrity_unverifiable_when_nothing_applies() {
let res = check_integrity(b"hello", Some(5), None, None, None, "b", "k").unwrap();
assert!(matches!(res, IntegrityCheck::Unverifiable));
}
#[test]
fn check_integrity_err_on_every_unsupported_checksum() {
for algo in [
ChecksumAlgorithm::Sha512,
ChecksumAlgorithm::Md5,
ChecksumAlgorithm::Xxhash64,
ChecksumAlgorithm::Xxhash3,
ChecksumAlgorithm::Xxhash128,
] {
let checksum = (algo.clone(), "anybase64value".to_string());
let err =
check_integrity(b"hello", None, None, None, Some(&checksum), "b", "k").unwrap_err();
assert!(
format!("{err:#}").contains("s3util cannot verify"),
"{algo:?}: unexpected error: {err:#}"
);
}
}
#[test]
fn check_integrity_err_on_content_length_mismatch() {
let err = check_integrity(b"hello", Some(999), None, None, None, "b", "k").unwrap_err();
assert!(format!("{err:#}").contains("content length mismatch"));
}
#[test]
fn check_integrity_err_on_etag_mismatch() {
let err = check_integrity(
b"hello",
None,
Some("\"00000000000000000000000000000000\""),
Some(&ServerSideEncryption::Aes256),
None,
"b",
"k",
)
.unwrap_err();
assert!(format!("{err:#}").contains("ETag (MD5) verification failed"));
}
#[test]
fn check_integrity_err_on_checksum_mismatch() {
let checksum = (ChecksumAlgorithm::Crc64Nvme, "AAAAAAAAAAA=".to_string());
let err = check_integrity(b"hello world", None, None, None, Some(&checksum), "b", "k")
.unwrap_err();
assert!(format!("{err:#}").contains("checksum verification failed"));
}
#[test]
fn verify_saved_file_ok_when_bytes_match_checksum() {
let payload = b"hello world";
let checksum = (
ChecksumAlgorithm::Crc64Nvme,
annotation::compute_checksum_base64(payload, ChecksumAlgorithm::Crc64Nvme),
);
let tmp = NamedTempFile::new().unwrap();
std::fs::write(tmp.path(), payload).unwrap();
let res = verify_saved_file(
tmp.path(),
Some(payload.len() as i64),
None,
None,
Some(&checksum),
"b",
"k",
);
assert!(res.is_ok(), "expected ok, got: {res:?}");
}
#[test]
fn verify_saved_file_err_when_bytes_corrupted() {
let payload = b"hello world";
let checksum = (
ChecksumAlgorithm::Crc64Nvme,
annotation::compute_checksum_base64(payload, ChecksumAlgorithm::Crc64Nvme),
);
let tmp = NamedTempFile::new().unwrap();
std::fs::write(tmp.path(), b"totally different bytes").unwrap();
let err =
verify_saved_file(tmp.path(), None, None, None, Some(&checksum), "b", "k").unwrap_err();
let msg = format!("{err:#}");
assert!(
msg.contains("corrupted while writing to disk"),
"got: {msg}"
);
}
#[test]
fn verify_saved_file_ok_when_nothing_verifiable() {
let payload = b"hello world";
let tmp = NamedTempFile::new().unwrap();
std::fs::write(tmp.path(), payload).unwrap();
let res = verify_saved_file(
tmp.path(),
Some(payload.len() as i64),
None,
None,
None,
"b",
"k",
);
assert!(res.is_ok(), "expected ok, got: {res:?}");
}
#[test]
fn verify_saved_file_err_when_file_unreadable() {
let dir = tempfile::tempdir().unwrap();
let missing = dir.path().join("never-written.bin");
let err = verify_saved_file(&missing, None, None, None, None, "b", "k").unwrap_err();
let msg = format!("{err:#}");
assert!(
msg.contains("re-reading the freshly-written temp file"),
"got: {msg}"
);
}
#[test]
fn write_verified_output_preserves_existing_file_on_verify_failure() {
let dir = tempfile::tempdir().unwrap();
let outfile = dir.path().join("out.bin");
std::fs::write(&outfile, b"PRE-EXISTING GOOD DATA").unwrap();
let payload = b"new payload bytes";
let wrong_checksum = (ChecksumAlgorithm::Crc64Nvme, "AAAAAAAAAAA=".to_string());
let err = write_verified_output(
outfile.to_str().unwrap(),
payload,
Some(payload.len() as i64),
None,
None,
Some(&wrong_checksum),
"b",
"k",
)
.unwrap_err();
assert!(
format!("{err:#}").contains("the destination was left untouched"),
"got: {err:#}"
);
assert_eq!(
std::fs::read(&outfile).unwrap(),
b"PRE-EXISTING GOOD DATA",
"pre-existing outfile must be preserved on verify failure"
);
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
1,
"only the untouched outfile should remain in the directory"
);
}
#[test]
fn write_verified_output_replaces_file_and_leaves_no_temp_on_success() {
let dir = tempfile::tempdir().unwrap();
let outfile = dir.path().join("out.bin");
std::fs::write(&outfile, b"OLD CONTENT").unwrap();
let payload = b"hello world";
let checksum = (
ChecksumAlgorithm::Crc64Nvme,
annotation::compute_checksum_base64(payload, ChecksumAlgorithm::Crc64Nvme),
);
write_verified_output(
outfile.to_str().unwrap(),
payload,
Some(payload.len() as i64),
None,
None,
Some(&checksum),
"b",
"k",
)
.unwrap();
assert_eq!(std::fs::read(&outfile).unwrap(), payload);
assert_eq!(
std::fs::read_dir(dir.path()).unwrap().count(),
1,
"temp file must be renamed onto the outfile, not left behind"
);
}
#[test]
fn write_verified_output_err_when_temp_file_cannot_be_created() {
let dir = tempfile::tempdir().unwrap();
let outfile = dir.path().join("no-such-subdir").join("out.bin");
let payload = b"payload bytes";
let err = write_verified_output(
outfile.to_str().unwrap(),
payload,
Some(payload.len() as i64),
None,
None,
None,
"b",
"k",
)
.unwrap_err();
assert!(
format!("{err:#}").contains("creating temp file next to"),
"got: {err:#}"
);
assert!(
!outfile.exists(),
"no file may be created when the temp write fails"
);
}
}