refactor(utils/os): Optimize Windows OS utilities and add safety comments (#1671)

Co-authored-by: weisd <weishidavip@163.com>

crates/e2e_test/Cargo.toml

@@ -36,7 +36,6 @@ serde_json.workspace = true
 tonic = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
-url.workspace = true
 rustfs-madmin.workspace = true
 rustfs-filemeta.workspace = true
 bytes.workspace = true
@@ -52,6 +51,7 @@ base64 = { workspace = true }
 rand = { workspace = true }
 chrono = { workspace = true }
 md5 = { workspace = true }
+sha2 = { workspace = true }
 suppaftp.workspace = true
 rcgen.workspace = true
 anyhow.workspace = true

crates/e2e_test/src/checksum_upload_test.rs

@@ -0,0 +1,229 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! E2E tests for PutObject and MultipartUpload with checksums (Content-MD5, x-amz-checksum-*).
+//! Verifies that uploads with Content-MD5 and x-amz-checksum-sha256 succeed and content is correct.
+
+#[cfg(test)]
+mod tests {
+    use crate::common::{RustFSTestEnvironment, init_logging};
+    use aws_sdk_s3::Client;
+    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
+    use base64::Engine;
+    use serial_test::serial;
+    use sha2::{Digest, Sha256};
+    use tracing::info;
+
+    fn create_s3_client(env: &RustFSTestEnvironment) -> Client {
+        env.create_s3_client()
+    }
+
+    async fn create_bucket(client: &Client, bucket: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        match client.create_bucket().bucket(bucket).send().await {
+            Ok(_) => {
+                info!("Bucket {} created successfully", bucket);
+                Ok(())
+            }
+            Err(e) => {
+                if e.to_string().contains("BucketAlreadyOwnedByYou") || e.to_string().contains("BucketAlreadyExists") {
+                    info!("Bucket {} already exists", bucket);
+                    Ok(())
+                } else {
+                    Err(Box::new(e))
+                }
+            }
+        }
+    }
+
+    fn content_md5_base64(body: &[u8]) -> String {
+        let digest = md5::compute(body);
+        base64::engine::general_purpose::STANDARD.encode(digest.as_slice())
+    }
+
+    fn checksum_sha256_base64(body: &[u8]) -> String {
+        let digest = Sha256::digest(body);
+        base64::engine::general_purpose::STANDARD.encode(digest.as_slice())
+    }
+
+    /// PutObject with Content-MD5: upload succeeds and GetObject returns same content.
+    #[tokio::test]
+    #[serial]
+    async fn test_put_object_with_content_md5() {
+        init_logging();
+        info!("TEST: PutObject with Content-MD5");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-checksum-md5";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        let key = "obj-with-md5.txt";
+        let content = b"Hello world with Content-MD5 checksum";
+        let content_md5 = content_md5_base64(content);
+
+        let result = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(content))
+            .content_md5(&content_md5)
+            .send()
+            .await;
+
+        assert!(result.is_ok(), "PutObject with Content-MD5 failed: {:?}", result.err());
+
+        let get_result = client.get_object().bucket(bucket).key(key).send().await;
+        assert!(get_result.is_ok(), "GetObject failed: {:?}", get_result.err());
+        let body_bytes = get_result.unwrap().body.collect().await.expect("collect body").into_bytes();
+        assert_eq!(body_bytes.as_ref(), content, "GetObject body must match uploaded content");
+        info!("PASSED: PutObject with Content-MD5 and GetObject content match");
+    }
+
+    /// PutObject with x-amz-checksum-sha256: upload succeeds and GetObject returns same content.
+    #[tokio::test]
+    #[serial]
+    async fn test_put_object_with_checksum_sha256() {
+        init_logging();
+        info!("TEST: PutObject with x-amz-checksum-sha256");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-checksum-sha256";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        let key = "obj-with-sha256.txt";
+        let content = b"Hello world with x-amz-checksum-sha256";
+        let checksum = checksum_sha256_base64(content);
+
+        let result = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(content))
+            .checksum_sha256(&checksum)
+            .send()
+            .await;
+
+        assert!(result.is_ok(), "PutObject with checksum_sha256 failed: {:?}", result.err());
+
+        let get_result = client.get_object().bucket(bucket).key(key).send().await;
+        assert!(get_result.is_ok(), "GetObject failed: {:?}", get_result.err());
+        let body_bytes = get_result.unwrap().body.collect().await.expect("collect body").into_bytes();
+        assert_eq!(body_bytes.as_ref(), content, "GetObject body must match uploaded content");
+        info!("PASSED: PutObject with checksum_sha256 and GetObject content match");
+    }
+
+    /// Multipart upload with checksum: CreateMultipartUpload, UploadPart(s) with checksum_sha256, CompleteMultipartUpload; then GetObject verifies content.
+    /// Uses part size >= 5MB (server minimum) for two parts.
+    #[tokio::test]
+    #[serial]
+    async fn test_multipart_upload_with_checksum() {
+        init_logging();
+        info!("TEST: MultipartUpload with checksum (checksum_sha256 on parts)");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-multipart-checksum";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        let key = "multipart-with-checksum.bin";
+        const PART_SIZE: usize = 6 * 1024 * 1024; // 6 MB per part (>= 5MB minimum)
+        let part1: Vec<u8> = (0..PART_SIZE).map(|i| (i % 256) as u8).collect();
+        let part2: Vec<u8> = (0..PART_SIZE).map(|i| ((i + 1) % 256) as u8).collect();
+        let full_content: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();
+
+        let create_result = client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256)
+            .send()
+            .await
+            .expect("Failed to create multipart upload");
+
+        let upload_id = create_result.upload_id().expect("No upload_id").to_string();
+
+        let checksum1 = checksum_sha256_base64(&part1);
+        let upload1 = client
+            .upload_part()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .part_number(1)
+            .body(ByteStream::from(part1.clone()))
+            .checksum_sha256(&checksum1)
+            .send()
+            .await
+            .expect("Failed to upload part 1");
+
+        let etag1 = upload1.e_tag().expect("No etag part 1").to_string();
+        let checksum_sha256_1 = upload1.checksum_sha256().map(|s| s.to_string());
+
+        let checksum2 = checksum_sha256_base64(&part2);
+        let upload2 = client
+            .upload_part()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .part_number(2)
+            .body(ByteStream::from(part2.clone()))
+            .checksum_sha256(&checksum2)
+            .send()
+            .await
+            .expect("Failed to upload part 2");
+
+        let etag2 = upload2.e_tag().expect("No etag part 2").to_string();
+        let checksum_sha256_2 = upload2.checksum_sha256().map(|s| s.to_string());
+
+        let mut part1_builder = CompletedPart::builder().part_number(1).e_tag(etag1);
+        if let Some(ref cs) = checksum_sha256_1 {
+            part1_builder = part1_builder.checksum_sha256(cs);
+        }
+        let mut part2_builder = CompletedPart::builder().part_number(2).e_tag(etag2);
+        if let Some(ref cs) = checksum_sha256_2 {
+            part2_builder = part2_builder.checksum_sha256(cs);
+        }
+
+        let completed_parts = vec![part1_builder.build(), part2_builder.build()];
+        let completed_upload = CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build();
+
+        let complete_result = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .multipart_upload(completed_upload)
+            .send()
+            .await;
+
+        assert!(complete_result.is_ok(), "CompleteMultipartUpload failed: {:?}", complete_result.err());
+
+        let get_result = client.get_object().bucket(bucket).key(key).send().await;
+        assert!(get_result.is_ok(), "GetObject failed: {:?}", get_result.err());
+        let body_bytes = get_result.unwrap().body.collect().await.expect("collect body").into_bytes();
+        assert_eq!(
+            body_bytes.as_ref(),
+            full_content.as_slice(),
+            "GetObject body must match concatenated parts"
+        );
+        info!("PASSED: MultipartUpload with checksum and GetObject content match");
+    }
+}

crates/e2e_test/src/lib.rs

@@ -60,3 +60,7 @@ mod protocols;
 // Object Lock tests
 #[cfg(test)]
 mod object_lock;
+
+// PutObject / MultipartUpload with checksum (Content-MD5, x-amz-checksum-*)
+#[cfg(test)]
+mod checksum_upload_test;

crates/ecstore/src/set_disk.rs

@@ -5730,16 +5730,11 @@ impl StorageAPI for SetDisks {
                 return Err(Error::other("checksum type not found"));
             };
 
-            if let Some(want) = &opts.want_checksum
-                && !want
-                    .checksum_type
-                    .is(rustfs_rio::ChecksumType::from_string_with_obj_type(cs, ct))
+            checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type(cs, ct);
+            if let Some(want) = opts.want_checksum.as_ref()
+                && !want.checksum_type.is(checksum_type)
             {
-                return Err(Error::other(format!(
-                    "checksum type mismatch, got {:?}, want {:?}",
-                    want,
-                    rustfs_rio::ChecksumType::from_string_with_obj_type(cs, ct)
-                )));
+                return Err(Error::other(format!("checksum type mismatch, got {:?}, want {:?}", want, checksum_type)));
             }
         }
 

crates/heal/Cargo.toml

@@ -30,30 +30,22 @@ categories = ["web-programming", "development-tools", "filesystem"]
 rustfs-config = { workspace = true }
 rustfs-ecstore = { workspace = true }
 rustfs-common = { workspace = true }
-rustfs-filemeta = { workspace = true }
 rustfs-madmin = { workspace = true }
 rustfs-utils = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
-time = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde"] }
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 futures = { workspace = true }
-s3s = { workspace = true }
-chrono = { workspace = true }
-rand = { workspace = true }
-reqwest = { workspace = true }
-tempfile = { workspace = true }
-walkdir = { workspace = true }
 
 [dev-dependencies]
 serde_json = { workspace = true }
 serial_test = { workspace = true }
 tracing-subscriber = { workspace = true }
 tempfile = { workspace = true }
-heed = { workspace = true }
+walkdir = { workspace = true }

crates/utils/src/os/linux.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{DiskInfo, IOStats};
+use crate::os::{DiskInfo, IOStats};
 use rustix::fs::statfs;
 use std::fs::File;
 use std::io::{self, BufRead, Error, ErrorKind};

crates/utils/src/os/unix.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{DiskInfo, IOStats};
+use crate::os::{DiskInfo, IOStats};
 use rustix::fs::{StatVfs, statvfs};
 use std::io::Error;
 use std::path::Path;

crates/utils/src/os/windows.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::{DiskInfo, IOStats};
+#![allow(unsafe_code)] // TODO: audit unsafe code
+
+use crate::os::{DiskInfo, IOStats};
 use std::io::Error;
 use std::path::Path;
 use windows::Win32::Foundation::MAX_PATH;
@@ -20,17 +22,15 @@ use windows::Win32::Storage::FileSystem::{GetDiskFreeSpaceExW, GetDiskFreeSpaceW
 
 /// Returns total and free bytes available in a directory, e.g. `C:\`.
 pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
-    let path_wide = p
-        .as_ref()
-        .to_string_lossy()
-        .encode_utf16()
-        .chain(std::iter::once(0))
-        .collect::<Vec<u16>>();
+    let path_wide = to_wide_path(p.as_ref());
 
     let mut free_bytes_available = 0u64;
     let mut total_number_of_bytes = 0u64;
     let mut total_number_of_free_bytes = 0u64;
 
+    // SAFETY:
+    // 1. `path_wide` is a valid null-terminated UTF-16 string.
+    // 2. Pointers to `u64` variables are valid and point to initialized stack memory.
     unsafe {
         GetDiskFreeSpaceExW(
             windows::core::PCWSTR::from_raw(path_wide.as_ptr()),
@@ -56,6 +56,9 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
     let mut number_of_free_clusters = 0u32;
     let mut total_number_of_clusters = 0u32;
 
+    // SAFETY:
+    // 1. `path_wide` is a valid null-terminated UTF-16 string.
+    // 2. Pointers to `u32` variables are valid and point to initialized stack memory.
     unsafe {
         GetDiskFreeSpaceW(
             windows::core::PCWSTR::from_raw(path_wide.as_ptr()),
@@ -87,6 +90,10 @@ fn get_windows_fs_type(p: &[u16]) -> std::io::Result<String> {
     let mut volume_name_buffer = [0u16; MAX_PATH as usize];
     let mut file_system_name_buffer = [0u16; MAX_PATH as usize];
 
+    // SAFETY:
+    // 1. `path` is a valid null-terminated UTF-16 string (volume root path).
+    // 2. Buffers are allocated with `MAX_PATH` size, which is sufficient for standard Windows paths.
+    // 3. Pointers to output variables are valid.
     unsafe {
         GetVolumeInformationW(
             windows::core::PCWSTR::from_raw(path.as_ptr()),
@@ -105,6 +112,11 @@ fn get_windows_fs_type(p: &[u16]) -> std::io::Result<String> {
 fn get_volume_name(v: &[u16]) -> std::io::Result<Vec<u16>> {
     let mut volume_name_buffer = [0u16; MAX_PATH as usize];
 
+    // SAFETY:
+    // 1. `v` is a valid null-terminated UTF-16 string.
+    // 2. `volume_name_buffer` is allocated with `MAX_PATH` size.
+    // 3. `GetVolumePathNameW` writes to the buffer and respects the buffer size (implicitly MAX_PATH for this API context usually, though explicit length param isn't present, it expects a buffer large enough).
+    // Note: GetVolumePathNameW documentation says "The buffer should be large enough to hold the path". MAX_PATH is generally safe for volume roots.
     unsafe {
         GetVolumePathNameW(windows::core::PCWSTR::from_raw(v.as_ptr()), &mut volume_name_buffer)
             .map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
@@ -122,9 +134,16 @@ fn utf16_to_string(v: &[u16]) -> String {
     String::from_utf16_lossy(&v[..len])
 }
 
+fn to_wide_path(path: &Path) -> Vec<u16> {
+    path.as_os_str().encode_wide().chain(std::iter::once(0)).collect()
+}
+
+// Helper trait to access encode_wide which is only available on Windows
+use std::os::windows::ffi::OsStrExt;
+
 pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
-    let path1_wide: Vec<u16> = disk1.encode_utf16().chain(std::iter::once(0)).collect();
-    let path2_wide: Vec<u16> = disk2.encode_utf16().chain(std::iter::once(0)).collect();
+    let path1_wide = to_wide_path(Path::new(disk1));
+    let path2_wide = to_wide_path(Path::new(disk2));
 
     let volume1 = get_volume_name(&path1_wide)?;
     let volume2 = get_volume_name(&path2_wide)?;