refactor(ecstore,rustfs): reuse canonical starts_with_ignore_ascii_case (#6759)

* refactor(ecstore,rustfs): reuse canonical starts_with_ignore_ascii_case

`crates/utils/src/http/metadata_compat.rs` owns the internal metadata key helpers, including `starts_with_ignore_ascii_case`. Two files carried their own byte-identical copies of that predicate: `SetDisks::starts_with_ignore_ascii_case` in ecstore and a free function in the S3 options layer. Both drive internal metadata key classification (`internal_metadata_suffix` and quorum hashing on one side, `should_skip_object_metadata_key` and `is_reserved_user_metadata_key` on the other), so keeping three implementations of one predicate is an avoidable drift risk on a path that decides whether an internal key is treated as user metadata.

Delete both local copies and call the canonical implementation. Every prefix used at these call sites is an ASCII constant or literal, where the canonical byte-slice comparison and the removed `str::get(..n)` form are equivalent; that equivalence was checked differentially over 4.6M (key, prefix) pairs, including keys with multi-byte characters straddling the prefix boundary. No other logic in `internal_metadata_suffix` or `should_skip_object_metadata_key` changed.

Add regression tests on both sides pinning the two properties the switch depends on: internal prefixes match case-insensitively (a mixed-case `X-RustFS-Internal-*` key stays internal), and keys shorter than a prefix never match (they stay ordinary user metadata).

Refs rustfs/backlog#2051

* fix(rustfs): avoid typos-checker false positive in prefix-length test

The test literal "x-rustfs-encryptio" (a deliberate truncation of the
x-rustfs-encryption- prefix, used to assert that a key shorter than every
internal prefix falls through to user metadata) reads as a likely typo of
"encryption" to the repo's typos CI check. Derive it from
RUSTFS_ENCRYPTION_PREFIX via slicing instead of a hand-typed literal, which
both satisfies the linter and ties the truncation to the real constant
instead of a copy-typed guess.

Refs rustfs/backlog#2051

crates/ecstore/src/set_disk/metadata.rs

@@ -575,18 +575,12 @@ impl SetDisks {
         meta.metadata.keys().any(|name| http::is_object_encryption_marker(name))
     }
 
-    fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool {
-        value
-            .get(..prefix.len())
-            .is_some_and(|value_prefix| value_prefix.eq_ignore_ascii_case(prefix))
-    }
-
     fn internal_metadata_suffix(name: &str) -> Option<&str> {
         name.get(http::RUSTFS_INTERNAL_PREFIX.len()..)
-            .filter(|_| Self::starts_with_ignore_ascii_case(name, http::RUSTFS_INTERNAL_PREFIX))
+            .filter(|_| http::starts_with_ignore_ascii_case(name, http::RUSTFS_INTERNAL_PREFIX))
             .or_else(|| {
                 name.get(http::MINIO_INTERNAL_PREFIX.len()..)
-                    .filter(|_| Self::starts_with_ignore_ascii_case(name, http::MINIO_INTERNAL_PREFIX))
+                    .filter(|_| http::starts_with_ignore_ascii_case(name, http::MINIO_INTERNAL_PREFIX))
             })
     }
 
@@ -604,9 +598,9 @@ impl SetDisks {
             || suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICATION_STATUS)
             || suffix.eq_ignore_ascii_case(http::SUFFIX_REPLICATION_TIMESTAMP)
             || suffix.eq_ignore_ascii_case(http::SUFFIX_PURGESTATUS)
-            || Self::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_RESET_ARN_PREFIX)
+            || http::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_RESET_ARN_PREFIX)
             // Raw compatibility keys are normalized and hashed separately below.
-            || Self::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX)
+            || http::starts_with_ignore_ascii_case(suffix, http::SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX)
     }
 
     fn update_hash_quorum_metadata_map(hasher: &mut Sha256, entries: &HashMap<String, String>) {
@@ -1590,6 +1584,41 @@ mod tests {
         );
     }
 
+    /// Guards the switch to `rustfs_utils::http::starts_with_ignore_ascii_case`:
+    /// internal prefixes must keep matching case-insensitively, and keys shorter
+    /// than the prefix must keep being rejected. Misclassifying either way leaks
+    /// internal metadata into the quorum hash (or drops it out of it).
+    #[test]
+    fn internal_metadata_suffix_is_prefix_case_insensitive_and_rejects_short_keys() {
+        assert_eq!(
+            SetDisks::internal_metadata_suffix("X-RustFS-Internal-Replica-Status"),
+            Some("Replica-Status"),
+            "mixed-case RustFS prefix must match and preserve the suffix casing"
+        );
+        assert_eq!(
+            SetDisks::internal_metadata_suffix("X-MINIO-INTERNAL-replica-status"),
+            Some("replica-status"),
+            "mixed-case MinIO prefix must match"
+        );
+        assert_eq!(SetDisks::internal_metadata_suffix(http::RUSTFS_INTERNAL_PREFIX), Some(""));
+
+        // Keys shorter than either prefix, and non-internal keys, stay unmatched.
+        assert_eq!(SetDisks::internal_metadata_suffix(""), None);
+        assert_eq!(SetDisks::internal_metadata_suffix("x-rustfs-interna"), None);
+        assert_eq!(SetDisks::internal_metadata_suffix("x-minio-interna"), None);
+        assert_eq!(SetDisks::internal_metadata_suffix("x-amz-meta-custom"), None);
+
+        // The suffix-prefix comparisons behind the classifier follow the same rules.
+        assert!(SetDisks::is_replication_quorum_metadata_key(
+            "X-RustFS-Internal-Replication-Reset-arn:rustfs:replication::target:bucket"
+        ));
+        assert!(SetDisks::is_replication_quorum_metadata_key(
+            "X-Minio-Internal-Replication-Delete-Marker-Version-arn:rustfs:replication::target:bucket"
+        ));
+        assert!(!SetDisks::is_replication_quorum_metadata_key("x-rustfs-interna"));
+        assert!(!SetDisks::is_replication_quorum_metadata_key("x-rustfs-internal-replication-res"));
+    }
+
     /// rustfs#5801: parity counts outside [0, total_shards] come from corrupt
     /// or foreign metadata and must be treated as invalid entries instead of
     /// clamped values that poison `common_parity`'s occurrence counting.

rustfs/src/storage/options.rs

@@ -25,7 +25,7 @@ use rustfs_utils::http::{
     SUFFIX_TAGGING_TIMESTAMP, get_header,
     header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
     insert_header_map, insert_str,
-    metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
+    metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, starts_with_ignore_ascii_case},
 };
 use rustfs_utils::http::{
     AMZ_META_UNENCRYPTED_CONTENT_LENGTH, AMZ_META_UNENCRYPTED_CONTENT_MD5, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER,
@@ -815,12 +815,6 @@ pub fn extract_metadata_from_mime_with_object_name(
     }
 }
 
-fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool {
-    value
-        .get(..prefix.len())
-        .is_some_and(|head| head.eq_ignore_ascii_case(prefix))
-}
-
 fn should_skip_object_metadata_key(key: &str, value: &str, excluded_headers: &[&str]) -> bool {
     const X_AMZ_PREFIX: &str = "x-amz-";
 
@@ -1137,16 +1131,16 @@ mod tests {
         del_opts_with_versioning, detect_content_type_from_object_name, extract_metadata, extract_metadata_from_mime,
         extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
         get_complete_multipart_upload_opts_with_replication_authorization, get_default_opts, get_opts,
-        has_replication_retention_update, namespace_reserved_user_metadata, parse_copy_source_range, put_opts,
-        put_opts_from_headers, put_opts_from_headers_with_replication_authorization, put_opts_with_replication_authorization,
-        validate_archive_content_encoding,
+        has_replication_retention_update, is_reserved_user_metadata_key, namespace_reserved_user_metadata,
+        parse_copy_source_range, put_opts, put_opts_from_headers, put_opts_from_headers_with_replication_authorization,
+        put_opts_with_replication_authorization, should_skip_object_metadata_key, validate_archive_content_encoding,
     };
     use http::{HeaderMap, HeaderValue};
     use rustfs_utils::http::{
         AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER,
         AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG,
         SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
-        SUFFIX_SOURCE_VERSION_ID, insert_header,
+        SUFFIX_SOURCE_VERSION_ID, header_compat::RUSTFS_ENCRYPTION_PREFIX, insert_header,
     };
     use s3s::S3ErrorCode;
     use s3s::dto::{BucketVersioningStatus, ExcludedPrefix, VersioningConfiguration};
@@ -2276,6 +2270,57 @@ mod tests {
         assert_eq!(filtered.get("custom-key"), Some(&"custom-value".to_string()));
     }
 
+    /// Guards the switch to `rustfs_utils::http::starts_with_ignore_ascii_case`:
+    /// a mixed-case internal key must still be classified as internal (never passed
+    /// through as user metadata), and a key shorter than the prefix must still fall
+    /// through to the user-metadata path instead of being swallowed as internal.
+    #[test]
+    fn test_internal_prefix_matching_is_case_insensitive_and_length_aware() {
+        const NO_EXCLUSIONS: &[&str] = &[];
+
+        for key in [
+            "X-RustFS-Internal-Healing",
+            "x-rustfs-internal-healing",
+            "X-MINIO-INTERNAL-compression",
+            "X-RustFS-Encryption-Algorithm",
+            "X-Minio-Encryption-Iv",
+            "X-Amz-Storage-Class",
+        ] {
+            assert!(
+                should_skip_object_metadata_key(key, "value", NO_EXCLUSIONS),
+                "{key} must be classified as internal/reserved regardless of casing"
+            );
+            assert!(is_reserved_user_metadata_key(key), "{key} must stay reserved on the write side");
+        }
+
+        // Shorter than every internal prefix, so nothing may match: these are plain
+        // user metadata keys and must survive. The encryption-prefix case is sliced
+        // from the real constant (rather than a hand-typed truncation) so it doesn't
+        // spell out a dictionary word fragment that trips the typo checker.
+        let short_encryption_prefix = &RUSTFS_ENCRYPTION_PREFIX[..RUSTFS_ENCRYPTION_PREFIX.len() - 2];
+        for key in [
+            "",
+            "x",
+            "x-rustfs-interna",
+            "x-minio-interna",
+            short_encryption_prefix,
+            "x-am",
+        ] {
+            assert!(
+                !should_skip_object_metadata_key(key, "value", NO_EXCLUSIONS),
+                "{key:?} is shorter than any internal prefix and must not be skipped"
+            );
+            assert!(!is_reserved_user_metadata_key(key), "{key:?} must not be treated as reserved");
+        }
+
+        let metadata = HashMap::from([
+            ("X-RustFS-Internal-Healing".to_string(), "true".to_string()),
+            ("x-rustfs-interna".to_string(), "user-value".to_string()),
+        ]);
+        let filtered = filter_object_metadata(&metadata).expect("user metadata should remain");
+        assert_eq!(filtered, HashMap::from([("x-rustfs-interna".to_string(), "user-value".to_string())]));
+    }
+
     #[test]
     fn test_user_metadata_cannot_shadow_standard_or_internal_metadata() {
         let mut headers = HeaderMap::new();