mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-02 11:29:17 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ed75e70a42 |
@@ -60,11 +60,6 @@ body-cache-whitelist-check: ## Check the body-cache eligibility gate stays a fai
|
||||
@echo "🧱 Checking body-cache whitelist guard..."
|
||||
./scripts/check_body_cache_whitelist.sh
|
||||
|
||||
.PHONY: fips-wording-check
|
||||
fips-wording-check: ## Check outward docs do not make unsupported FIPS claims
|
||||
@echo "📣 Checking FIPS wording guard..."
|
||||
./scripts/check_fips_wording.sh
|
||||
|
||||
.PHONY: log-analyzer-rules-check
|
||||
log-analyzer-rules-check: core-deps ## Check log-analyzer rule anchors still exist verbatim in source
|
||||
@echo "🩺 Checking log-analyzer rule anchors..."
|
||||
|
||||
@@ -19,13 +19,13 @@ planning-docs-check: ## Check that no planning-type documents are committed
|
||||
./scripts/check_no_planning_docs.sh
|
||||
|
||||
.PHONY: pre-commit
|
||||
pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
|
||||
pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
|
||||
@echo "✅ All pre-commit checks passed!"
|
||||
|
||||
.PHONY: pre-pr
|
||||
pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
|
||||
pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
|
||||
@echo "✅ All pre-PR checks passed!"
|
||||
|
||||
.PHONY: dev-check
|
||||
dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check fips-wording-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
|
||||
dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
|
||||
@echo "✅ Fast development checks passed!"
|
||||
|
||||
@@ -347,11 +347,6 @@ cargo run -p rustfs-filemeta --example dump_fileinfo -- "/path/to/file/xl.meta"
|
||||
absent, empty, and nil all mean "no value", never `Uuid::nil()`.
|
||||
- A remote-tier version of `None`/`""` means the tier bucket is unversioned:
|
||||
send **no** `versionId` on tier GET/DELETE.
|
||||
- Structs persisted in the scanner data-usage cache (`DataUsageCacheInfo`,
|
||||
`DataUsageEntry`) carry a hand-written map-encoded `Serialize`. MessagePack
|
||||
encodes derived structs as arrays, where an appended field makes the whole
|
||||
cache a decode error for older readers — keep new fields `#[serde(default)]`
|
||||
and keep the map encoding rather than reverting to `derive(Serialize)`.
|
||||
|
||||
## Naming Conventions
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use serde::{Deserialize, Serialize, ser::SerializeMap as _};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
hash::{DefaultHasher, Hash, Hasher},
|
||||
@@ -51,36 +51,24 @@ pub fn usage_last_update_is_untrusted_future(existing_last_update: SystemTime, n
|
||||
existing_last_update > now + USAGE_LAST_UPDATE_FUTURE_TOLERANCE
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)]
|
||||
pub struct TierStats {
|
||||
pub total_size: u64,
|
||||
pub num_versions: u64,
|
||||
pub num_objects: u64,
|
||||
pub num_versions: i32,
|
||||
pub num_objects: i32,
|
||||
}
|
||||
|
||||
impl TierStats {
|
||||
pub fn add(&self, u: &TierStats) -> TierStats {
|
||||
TierStats {
|
||||
total_size: self.total_size.saturating_add(u.total_size),
|
||||
num_versions: self.num_versions.saturating_add(u.num_versions),
|
||||
num_objects: self.num_objects.saturating_add(u.num_objects),
|
||||
total_size: self.total_size + u.total_size,
|
||||
num_versions: self.num_versions + u.num_versions,
|
||||
num_objects: self.num_objects + u.num_objects,
|
||||
}
|
||||
}
|
||||
|
||||
/// True when [`TierStats::add`] would report the exact sum instead of saturating.
|
||||
pub fn fits_add(&self, u: &TierStats) -> bool {
|
||||
self.total_size.checked_add(u.total_size).is_some()
|
||||
&& self.num_versions.checked_add(u.num_versions).is_some()
|
||||
&& self.num_objects.checked_add(u.num_objects).is_some()
|
||||
}
|
||||
|
||||
/// True when this tier contributed nothing, i.e. merging it is a no-op.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.total_size == 0 && self.num_versions == 0 && self.num_objects == 0
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
|
||||
pub struct AllTierStats {
|
||||
pub tiers: HashMap<String, TierStats>,
|
||||
}
|
||||
@@ -90,35 +78,31 @@ impl AllTierStats {
|
||||
Self { tiers: HashMap::new() }
|
||||
}
|
||||
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.tiers.is_empty()
|
||||
}
|
||||
|
||||
/// Folds a scan summary's per-tier map in.
|
||||
///
|
||||
/// Scanners seed the map with a zeroed entry for every configured tier, so
|
||||
/// empty contributions are skipped to keep the persisted cache from growing
|
||||
/// one key per tier on every folder that never held tiered data.
|
||||
pub fn add_sizes(&mut self, tiers: &HashMap<String, TierStats>) {
|
||||
pub fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
|
||||
for (tier, st) in tiers {
|
||||
if st.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let entry = self.tiers.entry(tier.clone()).or_default();
|
||||
*entry = entry.add(st);
|
||||
self.tiers
|
||||
.insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
|
||||
}
|
||||
}
|
||||
|
||||
pub fn merge(&mut self, other: &AllTierStats) {
|
||||
self.add_sizes(&other.tiers);
|
||||
pub fn merge(&mut self, other: AllTierStats) {
|
||||
for (tier, st) in other.tiers {
|
||||
self.tiers
|
||||
.insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
|
||||
}
|
||||
}
|
||||
|
||||
/// True when [`AllTierStats::merge`] would report exact sums for every tier.
|
||||
pub fn fits_merge(&self, other: &AllTierStats) -> bool {
|
||||
other
|
||||
.tiers
|
||||
.iter()
|
||||
.all(|(tier, right)| self.tiers.get(tier).is_none_or(|left| left.fits_add(right)))
|
||||
pub fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
|
||||
for (tier, st) in &self.tiers {
|
||||
stats.insert(
|
||||
tier.clone(),
|
||||
TierStats {
|
||||
total_size: st.total_size,
|
||||
num_versions: st.num_versions,
|
||||
num_objects: st.num_objects,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,14 +183,6 @@ pub struct DataUsageInfo {
|
||||
pub objects_total_size: u64,
|
||||
/// Replication info across all buckets
|
||||
pub replication_info: HashMap<String, BucketTargetUsageInfo>,
|
||||
/// Usage per storage class and remote tier across all buckets.
|
||||
///
|
||||
/// Absent on snapshots written before per-tier accounting was published,
|
||||
/// and on clusters with no remote tier configured: the scanner classifies
|
||||
/// objects by tier (including `STANDARD`/`REDUCED_REDUNDANCY`) only once a
|
||||
/// tier exists, so an absent value means "not accounted", never "zero".
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub tier_stats: Option<AllTierStats>,
|
||||
|
||||
/// Total number of buckets in this cluster
|
||||
pub buckets_count: u64,
|
||||
@@ -586,7 +562,7 @@ impl ReplicationAllStats {
|
||||
}
|
||||
|
||||
/// Data usage cache entry
|
||||
#[derive(Clone, Debug, Default, Deserialize)]
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
|
||||
pub struct DataUsageEntry {
|
||||
pub children: DataUsageHashMap,
|
||||
// These fields do not include any children.
|
||||
@@ -601,34 +577,6 @@ pub struct DataUsageEntry {
|
||||
/// Number of objects that failed to scan (e.g., IO errors)
|
||||
#[serde(default)]
|
||||
pub failed_objects: usize,
|
||||
/// Per-tier usage contributed by this entry, present only once a scan
|
||||
/// observed tier-classified objects.
|
||||
#[serde(default)]
|
||||
pub all_tier_stats: Option<AllTierStats>,
|
||||
}
|
||||
|
||||
impl Serialize for DataUsageEntry {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: serde::Serializer,
|
||||
{
|
||||
// Keep entries map-encoded so older readers can ignore fields appended
|
||||
// by newer scanner versions during rolling upgrades. The derived
|
||||
// (array) encoding made any appended field a decode error for them.
|
||||
let mut state = serializer.serialize_map(Some(11))?;
|
||||
state.serialize_entry("children", &self.children)?;
|
||||
state.serialize_entry("size", &self.size)?;
|
||||
state.serialize_entry("objects", &self.objects)?;
|
||||
state.serialize_entry("versions", &self.versions)?;
|
||||
state.serialize_entry("delete_markers", &self.delete_markers)?;
|
||||
state.serialize_entry("obj_sizes", &self.obj_sizes)?;
|
||||
state.serialize_entry("obj_versions", &self.obj_versions)?;
|
||||
state.serialize_entry("replication_stats", &self.replication_stats)?;
|
||||
state.serialize_entry("compacted", &self.compacted)?;
|
||||
state.serialize_entry("failed_objects", &self.failed_objects)?;
|
||||
state.serialize_entry("all_tier_stats", &self.all_tier_stats)?;
|
||||
state.end()
|
||||
}
|
||||
}
|
||||
|
||||
impl DataUsageEntry {
|
||||
@@ -687,22 +635,10 @@ impl DataUsageEntry {
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(o_tiers) = other.all_tier_stats.as_ref().filter(|tiers| !tiers.is_empty()) {
|
||||
self.all_tier_stats.get_or_insert_with(AllTierStats::new).merge(o_tiers);
|
||||
}
|
||||
|
||||
self.obj_sizes.merge_from(&other.obj_sizes);
|
||||
self.obj_versions.merge_from(&other.obj_versions);
|
||||
}
|
||||
|
||||
/// Folds a scan summary's per-tier map into this entry.
|
||||
pub fn add_tier_sizes(&mut self, tiers: &HashMap<String, TierStats>) {
|
||||
if tiers.values().all(TierStats::is_empty) {
|
||||
return;
|
||||
}
|
||||
self.all_tier_stats.get_or_insert_with(AllTierStats::new).add_sizes(tiers);
|
||||
}
|
||||
|
||||
pub fn checked_merge(&mut self, other: &DataUsageEntry) -> bool {
|
||||
let scalar_counts_fit = self.objects.checked_add(other.objects).is_some()
|
||||
&& self.versions.checked_add(other.versions).is_some()
|
||||
@@ -762,12 +698,7 @@ impl DataUsageEntry {
|
||||
}
|
||||
};
|
||||
|
||||
let tier_stats_fit = match (&self.all_tier_stats, &other.all_tier_stats) {
|
||||
(_, None) | (None, Some(_)) => true,
|
||||
(Some(left), Some(right)) => left.fits_merge(right),
|
||||
};
|
||||
|
||||
if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit {
|
||||
if !scalar_counts_fit || !histograms_fit || !replication_fits {
|
||||
return false;
|
||||
}
|
||||
self.merge(other);
|
||||
@@ -1107,7 +1038,6 @@ impl DataUsageCache {
|
||||
versions_total_count: flat.versions as u64,
|
||||
delete_markers_total_count: flat.delete_markers as u64,
|
||||
objects_total_size: flat.size as u64,
|
||||
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
|
||||
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
|
||||
buckets_usage,
|
||||
usage_snapshot_complete: self.info.snapshot_complete,
|
||||
@@ -1595,172 +1525,6 @@ mod tests {
|
||||
buckets_count: u64,
|
||||
}
|
||||
|
||||
fn tier_entry(tier: &str, stats: TierStats) -> DataUsageEntry {
|
||||
let mut entry = DataUsageEntry::default();
|
||||
entry.add_tier_sizes(&HashMap::from([(tier.to_string(), stats)]));
|
||||
entry
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tier_stats_survive_entry_merge() {
|
||||
let mut left = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 10,
|
||||
num_versions: 2,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
let mut right = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 5,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
right.add_tier_sizes(&HashMap::from([(
|
||||
"COLD".to_string(),
|
||||
TierStats {
|
||||
total_size: 7,
|
||||
num_versions: 1,
|
||||
num_objects: 0,
|
||||
},
|
||||
)]));
|
||||
|
||||
assert!(left.checked_merge(&right), "merging exact tier totals must be accepted");
|
||||
|
||||
let tiers = &left.all_tier_stats.expect("merged entry keeps tier stats").tiers;
|
||||
assert_eq!(
|
||||
tiers.get("WARM"),
|
||||
Some(&TierStats {
|
||||
total_size: 15,
|
||||
num_versions: 3,
|
||||
num_objects: 2,
|
||||
})
|
||||
);
|
||||
assert_eq!(
|
||||
tiers.get("COLD"),
|
||||
Some(&TierStats {
|
||||
total_size: 7,
|
||||
num_versions: 1,
|
||||
num_objects: 0,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tier_stats_merge_into_an_untiered_entry() {
|
||||
let mut left = DataUsageEntry::default();
|
||||
let right = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 10,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
|
||||
assert!(left.checked_merge(&right));
|
||||
|
||||
assert_eq!(
|
||||
left.all_tier_stats.expect("tier stats adopted from the merged entry").tiers["WARM"],
|
||||
TierStats {
|
||||
total_size: 10,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn checked_merge_rejects_overflowing_tier_totals() {
|
||||
let mut left = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: u64::MAX,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
let right = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 1,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
|
||||
assert!(!left.checked_merge(&right), "saturating tier totals must not be published");
|
||||
assert_eq!(left.all_tier_stats.expect("left is untouched").tiers["WARM"].total_size, u64::MAX);
|
||||
}
|
||||
|
||||
/// Entry shape released before per-tier accounting, using the derived
|
||||
/// (array) encoding those writers produced.
|
||||
#[derive(Serialize, Deserialize)]
|
||||
struct LegacyEntry {
|
||||
children: DataUsageHashMap,
|
||||
size: usize,
|
||||
objects: usize,
|
||||
versions: usize,
|
||||
delete_markers: usize,
|
||||
obj_sizes: SizeHistogram,
|
||||
obj_versions: VersionsHistogram,
|
||||
replication_stats: Option<ReplicationAllStats>,
|
||||
compacted: bool,
|
||||
#[serde(default)]
|
||||
failed_objects: usize,
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn entries_are_map_encoded_so_appended_fields_stay_readable() {
|
||||
// A derived (array) encoding turns every appended field into a decode
|
||||
// error for readers built before it existed, which would cost a mixed
|
||||
// -version cluster its whole scan cache. Entries must stay map-encoded.
|
||||
let current = tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 3,
|
||||
num_versions: 1,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
let mut encoded = Vec::new();
|
||||
current
|
||||
.serialize(&mut rmp_serde::Serializer::new(&mut encoded))
|
||||
.expect("encode current entry");
|
||||
|
||||
let legacy: LegacyEntry = rmp_serde::from_slice(&encoded).expect("legacy reader should ignore the appended field");
|
||||
assert_eq!(legacy.objects, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_array_encoded_entries_still_load() {
|
||||
let legacy = LegacyEntry {
|
||||
children: DataUsageHashMap::default(),
|
||||
size: 12,
|
||||
objects: 3,
|
||||
versions: 4,
|
||||
delete_markers: 1,
|
||||
obj_sizes: SizeHistogram::default(),
|
||||
obj_versions: VersionsHistogram::default(),
|
||||
replication_stats: None,
|
||||
compacted: false,
|
||||
failed_objects: 2,
|
||||
};
|
||||
let mut encoded = Vec::new();
|
||||
legacy
|
||||
.serialize(&mut rmp_serde::Serializer::new(&mut encoded))
|
||||
.expect("encode legacy entry");
|
||||
|
||||
let decoded: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("current reader should default the missing field");
|
||||
|
||||
assert_eq!(decoded.size, 12);
|
||||
assert_eq!(decoded.failed_objects, 2);
|
||||
assert!(decoded.all_tier_stats.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hash_path_uses_portable_slash_semantics() {
|
||||
for (input, expected) in [
|
||||
@@ -2137,44 +1901,6 @@ mod tests {
|
||||
assert_eq!(info.buckets_count, 2);
|
||||
assert!(info.buckets_usage.is_empty());
|
||||
assert_eq!(info.objects_total_count, 3);
|
||||
assert!(info.tier_stats.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_dui_reports_tier_usage_from_the_flattened_tree() {
|
||||
let root_hash = hash_path("root");
|
||||
let bucket_hash = hash_path("bucket-a");
|
||||
let mut cache = DataUsageCache {
|
||||
info: DataUsageCacheInfo {
|
||||
name: "root".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
cache.replace_hashed(&root_hash, &None, &DataUsageEntry::default());
|
||||
cache.replace_hashed(
|
||||
&bucket_hash,
|
||||
&Some(root_hash),
|
||||
&tier_entry(
|
||||
"WARM",
|
||||
TierStats {
|
||||
total_size: 40,
|
||||
num_versions: 2,
|
||||
num_objects: 2,
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
let info = cache.dui("root", &["bucket-a".to_string()]);
|
||||
|
||||
assert_eq!(
|
||||
info.tier_stats.expect("child tier usage should roll up to the root").tiers["WARM"],
|
||||
TierStats {
|
||||
total_size: 40,
|
||||
num_versions: 2,
|
||||
num_objects: 2,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -24,11 +24,10 @@ use rustfs_replication::{
|
||||
};
|
||||
use rustfs_utils::http::{
|
||||
AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
|
||||
AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID,
|
||||
AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE,
|
||||
HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
|
||||
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map,
|
||||
is_internal_key,
|
||||
AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT,
|
||||
CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, HeaderExt as _,
|
||||
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE,
|
||||
SUFFIX_REPLICATION_SSEC_CRC, SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map, is_internal_key,
|
||||
};
|
||||
use time::OffsetDateTime;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
@@ -80,48 +79,6 @@ static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[
|
||||
];
|
||||
|
||||
const ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED: &str = "managed SSE replication requires target encryption support";
|
||||
const ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED: &str = "replication source contains unsupported encryption metadata";
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum ReplicationSourceEncryption {
|
||||
Plaintext,
|
||||
SseS3,
|
||||
SseKms,
|
||||
SseC,
|
||||
Unsupported,
|
||||
}
|
||||
|
||||
fn metadata_value<'a>(metadata: &'a HashMap<String, String>, name: &str) -> Option<&'a str> {
|
||||
metadata
|
||||
.iter()
|
||||
.find(|(key, _)| key.eq_ignore_ascii_case(name))
|
||||
.map(|(_, value)| value.as_str())
|
||||
}
|
||||
|
||||
fn classify_replication_source_encryption(metadata: &HashMap<String, String>) -> ReplicationSourceEncryption {
|
||||
let is_ssec = replication_object_is_ssec_encrypted(metadata);
|
||||
let sse = metadata_value(metadata, AMZ_SERVER_SIDE_ENCRYPTION);
|
||||
let kms_key_id = metadata_value(metadata, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID);
|
||||
let kms_context = metadata_value(metadata, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT);
|
||||
|
||||
if is_ssec {
|
||||
return if sse.is_some() || kms_key_id.is_some() || kms_context.is_some() {
|
||||
ReplicationSourceEncryption::Unsupported
|
||||
} else {
|
||||
ReplicationSourceEncryption::SseC
|
||||
};
|
||||
}
|
||||
|
||||
match sse.map(str::trim) {
|
||||
None if kms_key_id.is_none() && kms_context.is_none() => ReplicationSourceEncryption::Plaintext,
|
||||
Some(value) if value.eq_ignore_ascii_case("AES256") && kms_key_id.is_none() && kms_context.is_none() => {
|
||||
ReplicationSourceEncryption::SseS3
|
||||
}
|
||||
Some(value) if value.eq_ignore_ascii_case("aws:kms") => ReplicationSourceEncryption::SseKms,
|
||||
_ if kms_key_id.is_some() => ReplicationSourceEncryption::SseKms,
|
||||
_ => ReplicationSourceEncryption::Unsupported,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String, String>) -> bool {
|
||||
rustfs_replication::is_ssec_encrypted(user_defined)
|
||||
@@ -152,18 +109,7 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
|
||||
use rustfs_utils::http::{AMZ_CHECKSUM_TYPE, AMZ_CHECKSUM_TYPE_FULL_OBJECT};
|
||||
|
||||
let mut meta = HashMap::new();
|
||||
let source_encryption = classify_replication_source_encryption(&object_info.user_defined);
|
||||
let is_ssec = matches!(source_encryption, ReplicationSourceEncryption::SseC);
|
||||
|
||||
match source_encryption {
|
||||
ReplicationSourceEncryption::Plaintext | ReplicationSourceEncryption::SseC => {}
|
||||
ReplicationSourceEncryption::SseS3 | ReplicationSourceEncryption::SseKms => {
|
||||
return Err(Error::other(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED));
|
||||
}
|
||||
ReplicationSourceEncryption::Unsupported => {
|
||||
return Err(Error::other(ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED));
|
||||
}
|
||||
}
|
||||
let is_ssec = replication_object_is_ssec_encrypted(&object_info.user_defined);
|
||||
|
||||
for (key, value) in object_info.user_defined.iter() {
|
||||
let has_valid_sse_header = valid_sse_replication_header(key).is_some();
|
||||
@@ -289,6 +235,20 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
|
||||
};
|
||||
}
|
||||
|
||||
let has_sse_s3 = object_info
|
||||
.user_defined
|
||||
.get(AMZ_SERVER_SIDE_ENCRYPTION)
|
||||
.is_some_and(|value| value.eq_ignore_ascii_case("AES256"));
|
||||
let has_sse_kms = object_info
|
||||
.user_defined
|
||||
.get(AMZ_SERVER_SIDE_ENCRYPTION)
|
||||
.is_some_and(|value| value.eq_ignore_ascii_case("aws:kms"))
|
||||
|| object_info.user_defined.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID);
|
||||
|
||||
if has_sse_s3 || has_sse_kms {
|
||||
return Err(Error::other(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED));
|
||||
}
|
||||
|
||||
Ok((put_options, is_multipart))
|
||||
}
|
||||
|
||||
@@ -626,46 +586,6 @@ mod tests {
|
||||
assert!(get_header_map(&options.user_metadata, SUFFIX_REPLICATION_SSEC_CRC).is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replication_source_encryption_classification_is_explicit_and_fail_closed() {
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::new()),
|
||||
ReplicationSourceEncryption::Plaintext
|
||||
);
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::from([(
|
||||
"x-amz-server-side-encryption".to_string(),
|
||||
"AES256".to_string()
|
||||
)])),
|
||||
ReplicationSourceEncryption::SseS3
|
||||
);
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::from([(
|
||||
"x-amz-server-side-encryption".to_string(),
|
||||
"AWS:KMS".to_string()
|
||||
)])),
|
||||
ReplicationSourceEncryption::SseKms
|
||||
);
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())])),
|
||||
ReplicationSourceEncryption::SseC
|
||||
);
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::from([(
|
||||
"x-amz-server-side-encryption".to_string(),
|
||||
"unsupported-algorithm".to_string(),
|
||||
)])),
|
||||
ReplicationSourceEncryption::Unsupported
|
||||
);
|
||||
assert_eq!(
|
||||
classify_replication_source_encryption(&HashMap::from([(
|
||||
AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT.to_string(),
|
||||
"opaque-context".to_string(),
|
||||
)])),
|
||||
ReplicationSourceEncryption::Unsupported
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replication_put_options_rejects_sse_s3_until_target_encryption_is_supported() {
|
||||
let object_info = ObjectInfo {
|
||||
@@ -699,22 +619,6 @@ mod tests {
|
||||
assert!(err.to_string().contains(ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replication_put_options_rejects_unknown_encryption_without_echoing_metadata() {
|
||||
let secret_like_value = "opaque-context-that-must-not-be-logged";
|
||||
let object_info = ObjectInfo {
|
||||
user_defined: Arc::new(HashMap::from([
|
||||
(AMZ_SERVER_SIDE_ENCRYPTION.to_string(), "unsupported-algorithm".to_string()),
|
||||
(AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT.to_string(), secret_like_value.to_string()),
|
||||
])),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let err = replication_put_object_options("", &object_info).expect_err("unknown encryption must fail closed");
|
||||
assert!(err.to_string().contains(ERR_REPLICATION_ENCRYPTION_METADATA_UNSUPPORTED));
|
||||
assert!(!err.to_string().contains(secret_like_value));
|
||||
}
|
||||
|
||||
// T3 (#1264): the outbound replication path forwards a stored object checksum into
|
||||
// user_metadata via decrypt_checksums, which is algorithm-agnostic. This locks that
|
||||
// the AWS 2026-04 additional algorithms (XXHash3/64/128, SHA-512, MD5) are forwarded
|
||||
|
||||
@@ -104,12 +104,21 @@ pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -
|
||||
if rule.destination.encryption_configuration.is_some() {
|
||||
return Some("Destination.EncryptionConfiguration");
|
||||
}
|
||||
if rule.destination.access_control_translation.is_some() {
|
||||
return Some("Destination.AccessControlTranslation");
|
||||
}
|
||||
if rule.destination.account.is_some() {
|
||||
return Some("Destination.Account");
|
||||
}
|
||||
if rule.destination.metrics.is_some() {
|
||||
return Some("Destination.Metrics");
|
||||
}
|
||||
if rule.destination.replication_time.is_some() {
|
||||
return Some("Destination.ReplicationTime");
|
||||
}
|
||||
if rule.destination.storage_class.is_some() {
|
||||
return Some("Destination.StorageClass");
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
@@ -422,6 +431,7 @@ mod tests {
|
||||
MetricsStatus, ReplicaModifications, ReplicationRule, ReplicationTime, ReplicationTimeStatus, ReplicationTimeValue,
|
||||
SourceSelectionCriteria, SseKmsEncryptedObjects, SseKmsEncryptedObjectsStatus,
|
||||
};
|
||||
use s3s::xml::{Deserializer, Serializer};
|
||||
|
||||
fn replication_rule(id: &str, arn: &str) -> ReplicationRule {
|
||||
ReplicationRule {
|
||||
@@ -813,12 +823,23 @@ mod tests {
|
||||
);
|
||||
|
||||
config.rules[0].source_selection_criteria = None;
|
||||
config.rules[0].destination.encryption_configuration = Some(EncryptionConfiguration {
|
||||
replica_kms_key_id: Some("arn:aws:kms:us-east-1:123456789012:key/opaque-key-id".to_string()),
|
||||
});
|
||||
config.rules[0].destination.encryption_configuration = Some(EncryptionConfiguration::default());
|
||||
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.EncryptionConfiguration"));
|
||||
|
||||
config.rules[0].destination.encryption_configuration = None;
|
||||
config.rules[0].destination.access_control_translation = Some(s3s::dto::AccessControlTranslation {
|
||||
owner: s3s::dto::OwnerOverride::from_static(s3s::dto::OwnerOverride::DESTINATION),
|
||||
});
|
||||
assert_eq!(
|
||||
unsupported_replication_config_field(&config),
|
||||
Some("Destination.AccessControlTranslation")
|
||||
);
|
||||
|
||||
config.rules[0].destination.access_control_translation = None;
|
||||
config.rules[0].destination.account = Some("123456789012".to_string());
|
||||
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.Account"));
|
||||
|
||||
config.rules[0].destination.account = None;
|
||||
config.rules[0].destination.metrics = Some(Metrics {
|
||||
event_threshold: None,
|
||||
status: MetricsStatus::from_static(MetricsStatus::ENABLED),
|
||||
@@ -831,6 +852,72 @@ mod tests {
|
||||
time: ReplicationTimeValue { minutes: Some(15) },
|
||||
});
|
||||
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime"));
|
||||
|
||||
config.rules[0].destination.replication_time = None;
|
||||
config.rules[0].destination.storage_class =
|
||||
Some(s3s::dto::StorageClass::from_static(s3s::dto::StorageClass::STANDARD_IA));
|
||||
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.StorageClass"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn historical_destination_fields_survive_the_s3_xml_round_trip() {
|
||||
let xml = br#"
|
||||
<ReplicationConfiguration>
|
||||
<Role></Role>
|
||||
<Rule>
|
||||
<ID>historical</ID>
|
||||
<Status>Enabled</Status>
|
||||
<Destination>
|
||||
<Bucket>arn:aws:s3:::destination</Bucket>
|
||||
<Account>123456789012</Account>
|
||||
<AccessControlTranslation><Owner>Destination</Owner></AccessControlTranslation>
|
||||
<StorageClass>STANDARD_IA</StorageClass>
|
||||
</Destination>
|
||||
</Rule>
|
||||
</ReplicationConfiguration>
|
||||
"#;
|
||||
let mut deserializer = Deserializer::new(xml);
|
||||
let config = <ReplicationConfiguration as s3s::xml::Deserialize>::deserialize(&mut deserializer)
|
||||
.expect("historical config should parse");
|
||||
deserializer
|
||||
.expect_eof()
|
||||
.expect("historical config should consume the whole body");
|
||||
|
||||
let mut encoded = Vec::new();
|
||||
<ReplicationConfiguration as s3s::xml::Serialize>::serialize(&config, &mut Serializer::new(&mut encoded))
|
||||
.expect("historical config should serialize");
|
||||
let encoded = String::from_utf8(encoded).expect("serialized XML should be UTF-8");
|
||||
for field in [
|
||||
"<Account>123456789012</Account>",
|
||||
"<Owner>Destination</Owner>",
|
||||
"<StorageClass>STANDARD_IA</StorageClass>",
|
||||
] {
|
||||
assert!(encoded.contains(field), "historical field {field} was lost: {encoded}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn s3_xml_parser_discards_unknown_replication_elements_before_validation() {
|
||||
let xml = br#"
|
||||
<ReplicationConfiguration>
|
||||
<Role></Role>
|
||||
<FutureTopLevel>future</FutureTopLevel>
|
||||
<Rule>
|
||||
<ID>unknown</ID>
|
||||
<Status>Enabled</Status>
|
||||
<Destination>
|
||||
<Bucket>arn:aws:s3:::destination</Bucket>
|
||||
</Destination>
|
||||
</Rule>
|
||||
</ReplicationConfiguration>
|
||||
"#;
|
||||
let mut deserializer = Deserializer::new(xml);
|
||||
let config = <ReplicationConfiguration as s3s::xml::Deserialize>::deserialize(&mut deserializer)
|
||||
.expect("s3s should accept unknown elements");
|
||||
deserializer.expect_eof().expect("unknown elements should still be consumed");
|
||||
|
||||
assert!(config.rules[0].destination.encryption_configuration.is_none());
|
||||
assert_eq!(unsupported_replication_config_field(&config), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -27,8 +27,8 @@ use rustfs_common::heal_channel::HealScanMode;
|
||||
#[cfg(test)]
|
||||
use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
|
||||
pub use rustfs_data_usage::{
|
||||
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash,
|
||||
DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path,
|
||||
BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash, DataUsageHashMap,
|
||||
DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, hash_path,
|
||||
};
|
||||
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
||||
use tokio::time::{Duration, Instant, sleep, timeout};
|
||||
@@ -184,6 +184,69 @@ pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
|
||||
|
||||
const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
|
||||
|
||||
#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)]
|
||||
pub struct TierStats {
|
||||
pub total_size: u64,
|
||||
pub num_versions: i32,
|
||||
pub num_objects: i32,
|
||||
}
|
||||
|
||||
impl TierStats {
|
||||
pub fn add(&self, u: &TierStats) -> TierStats {
|
||||
TierStats {
|
||||
total_size: self.total_size + u.total_size,
|
||||
num_versions: self.num_versions + u.num_versions,
|
||||
num_objects: self.num_objects + u.num_objects,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_object_info(oi: &ObjectInfo) -> Self {
|
||||
TierStats {
|
||||
total_size: oi.size as u64,
|
||||
num_versions: 1,
|
||||
num_objects: if oi.is_latest { 1 } else { 0 },
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
|
||||
pub struct AllTierStats {
|
||||
pub tiers: HashMap<String, TierStats>,
|
||||
}
|
||||
|
||||
impl AllTierStats {
|
||||
pub fn new() -> Self {
|
||||
Self { tiers: HashMap::new() }
|
||||
}
|
||||
|
||||
pub fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
|
||||
for (tier, st) in tiers {
|
||||
self.tiers
|
||||
.insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
|
||||
}
|
||||
}
|
||||
|
||||
pub fn merge(&mut self, other: AllTierStats) {
|
||||
for (tier, st) in other.tiers {
|
||||
self.tiers
|
||||
.insert(tier.clone(), self.tiers.get(&tier).copied().unwrap_or_default().add(&st));
|
||||
}
|
||||
}
|
||||
|
||||
pub fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
|
||||
for (tier, st) in &self.tiers {
|
||||
stats.insert(
|
||||
tier.clone(),
|
||||
TierStats {
|
||||
total_size: st.total_size,
|
||||
num_versions: st.num_versions,
|
||||
num_objects: st.num_objects,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Size summary for a single object or group of objects
|
||||
#[derive(Debug, Default, Clone)]
|
||||
pub struct SizeSummary {
|
||||
@@ -238,11 +301,7 @@ impl SizeSummary {
|
||||
}
|
||||
|
||||
if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
|
||||
*tier_stats = tier_stats.add(&TierStats {
|
||||
total_size: u64::try_from(oi.size).unwrap_or(0),
|
||||
num_versions: 1,
|
||||
num_objects: u64::from(oi.is_latest),
|
||||
});
|
||||
*tier_stats = tier_stats.add(&TierStats::from_object_info(oi));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -904,7 +963,6 @@ impl DataUsageCache {
|
||||
versions_total_count: flat.versions as u64,
|
||||
delete_markers_total_count: flat.delete_markers as u64,
|
||||
objects_total_size: flat.size as u64,
|
||||
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
|
||||
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
|
||||
buckets_usage,
|
||||
..Default::default()
|
||||
|
||||
@@ -461,8 +461,6 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
|
||||
.pending_count
|
||||
.saturating_add(u64::try_from(st.pending_count).unwrap_or(u64::MAX));
|
||||
}
|
||||
|
||||
into.add_tier_sizes(&summary.tier_stats);
|
||||
}
|
||||
|
||||
/// Cached folder information for scanning
|
||||
@@ -3051,7 +3049,7 @@ mod tests {
|
||||
|
||||
use super::*;
|
||||
use crate::storage_api::VersionPurgeStatusType;
|
||||
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
|
||||
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, new_disk};
|
||||
use rustfs_filemeta::{FileInfo, FileMeta};
|
||||
use serial_test::serial;
|
||||
#[cfg(unix)]
|
||||
@@ -3130,56 +3128,6 @@ mod tests {
|
||||
assert_eq!(target_stats.pending_count, u64::MAX);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_size_summary_application_accumulates_tier_stats() {
|
||||
let mut entry = DataUsageEntry::default();
|
||||
let mut summary = SizeSummary::default();
|
||||
summary.tier_stats.insert(
|
||||
"WARM".to_string(),
|
||||
TierStats {
|
||||
total_size: 100,
|
||||
num_versions: 2,
|
||||
num_objects: 1,
|
||||
},
|
||||
);
|
||||
// Scanners seed a zeroed entry for every configured tier; those must not
|
||||
// reach the cache as empty keys.
|
||||
summary
|
||||
.tier_stats
|
||||
.insert(storageclass::STANDARD.to_string(), TierStats::default());
|
||||
|
||||
apply_scanner_size_summary(&mut entry, &summary);
|
||||
apply_scanner_size_summary(&mut entry, &summary);
|
||||
|
||||
let tiers = entry.all_tier_stats.as_ref().expect("tier stats should be recorded");
|
||||
assert_eq!(
|
||||
tiers.tiers.get("WARM"),
|
||||
Some(&TierStats {
|
||||
total_size: 200,
|
||||
num_versions: 4,
|
||||
num_objects: 2,
|
||||
})
|
||||
);
|
||||
assert!(!tiers.tiers.contains_key(storageclass::STANDARD));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_size_summary_application_skips_untiered_summaries() {
|
||||
let mut entry = DataUsageEntry::default();
|
||||
let mut summary = SizeSummary {
|
||||
total_size: 10,
|
||||
..Default::default()
|
||||
};
|
||||
summary
|
||||
.tier_stats
|
||||
.insert(storageclass::STANDARD.to_string(), TierStats::default());
|
||||
summary.tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
|
||||
|
||||
apply_scanner_size_summary(&mut entry, &summary);
|
||||
|
||||
assert!(entry.all_tier_stats.is_none(), "zero-only tier maps must not allocate cache state");
|
||||
}
|
||||
|
||||
async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
|
||||
let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4()));
|
||||
tokio::fs::create_dir_all(&temp_dir)
|
||||
|
||||
@@ -1146,7 +1146,6 @@ fn completed_data_usage_info(
|
||||
versions_total_count: u64::try_from(total.versions).ok()?,
|
||||
delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
|
||||
objects_total_size: u64::try_from(total.size).ok()?,
|
||||
tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
|
||||
buckets_count: u64::try_from(all_buckets.len()).ok()?,
|
||||
bucket_sizes,
|
||||
buckets_usage,
|
||||
@@ -1251,57 +1250,6 @@ mod publish_gate_tests {
|
||||
completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_publishes_tier_stats_across_sets() {
|
||||
let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];
|
||||
let warm = |total_size, num_versions, num_objects| {
|
||||
HashMap::from([(
|
||||
"WARM".to_string(),
|
||||
TierStats {
|
||||
total_size,
|
||||
num_versions,
|
||||
num_objects,
|
||||
},
|
||||
)])
|
||||
};
|
||||
|
||||
let mut first_set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
|
||||
let mut tiered = DataUsageEntry::default();
|
||||
tiered.add_tier_sizes(&warm(100, 2, 1));
|
||||
first_set.replace("bucket-b", DATA_USAGE_ROOT, tiered);
|
||||
|
||||
let mut second_set = completed_root_cache("bucket-b", 2, 20, DataUsageCacheSource::new(1, 0));
|
||||
let mut tiered = DataUsageEntry::default();
|
||||
tiered.add_tier_sizes(&warm(50, 1, 1));
|
||||
second_set.replace("bucket-a", DATA_USAGE_ROOT, tiered);
|
||||
|
||||
let (data_usage_info, _) = completed_data_usage_info_for_test(&[first_set, second_set], &all_buckets, false, false)
|
||||
.expect("completed sets should publish a snapshot");
|
||||
|
||||
assert_eq!(
|
||||
data_usage_info
|
||||
.tier_stats
|
||||
.expect("tier usage should reach the snapshot")
|
||||
.tiers["WARM"],
|
||||
TierStats {
|
||||
total_size: 150,
|
||||
num_versions: 3,
|
||||
num_objects: 2,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_omits_tier_stats_without_tiered_objects() {
|
||||
let all_buckets = vec!["bucket-a".to_string()];
|
||||
let set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
|
||||
|
||||
let (data_usage_info, _) = completed_data_usage_info_for_test(&[set], &all_buckets, false, false)
|
||||
.expect("completed set should publish a snapshot");
|
||||
|
||||
assert!(data_usage_info.tier_stats.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_requires_every_set_before_publish() {
|
||||
let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string(), "bucket-empty".to_string()];
|
||||
|
||||
@@ -51,7 +51,7 @@ Suggested boilerplate when the topic cannot be avoided:
|
||||
|
||||
### Guard
|
||||
|
||||
`README.md` and `CHANGELOG.md` currently contain no FIPS-related wording; `scripts/check_fips_wording.sh` is the grep guard for that public baseline. Any future occurrence of the banned strings in either file should be treated as a defect and either removed or brought under the qualifier rule above. This document intentionally contains the terminology needed to define the policy and is not part of that narrow outward-material scan.
|
||||
There is currently no FIPS-related wording anywhere in the repository's Markdown; that clean baseline is what a grep anchor test protects. Any future occurrence of the banned strings in shipped documentation should be treated as a defect and either removed or brought under the qualifier rule above.
|
||||
|
||||
## The `rustfs-crypto` `fips` feature: what it actually does
|
||||
|
||||
|
||||
@@ -34,7 +34,6 @@ use rustfs_targets::{
|
||||
manifest::builtin_target_manifest,
|
||||
target::{TargetType, mqtt::MQTTTlsConfig},
|
||||
};
|
||||
use rustfs_utils::egress::{ENV_OUTBOUND_ALLOW_ORIGINS, OutboundPolicy};
|
||||
use s3s::{Body, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
|
||||
use serde::Serialize;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
@@ -706,18 +705,6 @@ async fn validate_webhook_request(kv_map: &HashMap<String, String>) -> S3Result<
|
||||
));
|
||||
}
|
||||
}
|
||||
let outbound_policy =
|
||||
OutboundPolicy::from_env_cached().map_err(|e| s3_error!(InvalidArgument, "invalid outbound policy: {}", e))?;
|
||||
outbound_policy.validate_url(&parsed_endpoint).map_err(|e| {
|
||||
s3_error!(
|
||||
InvalidArgument,
|
||||
"endpoint is not allowed by the outbound policy: {}; review {}; private origins must be listed as exact origins such as {}={} and RustFS restarted (metadata and link-local destinations remain blocked)",
|
||||
e,
|
||||
ENV_OUTBOUND_ALLOW_ORIGINS,
|
||||
ENV_OUTBOUND_ALLOW_ORIGINS,
|
||||
parsed_endpoint.origin().ascii_serialization()
|
||||
)
|
||||
})?;
|
||||
if let Some(queue_dir) = kv_map.get("queue_dir") {
|
||||
validate_queue_dir(queue_dir.as_str()).await?;
|
||||
}
|
||||
@@ -966,34 +953,3 @@ fn to_kvs(kv_map: &HashMap<String, String>) -> rustfs_config::server_config::KVS
|
||||
}
|
||||
kvs
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod webhook_request_tests {
|
||||
use super::validate_webhook_request;
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[tokio::test]
|
||||
async fn private_webhook_request_is_rejected_with_operator_action() {
|
||||
let config = HashMap::from([("endpoint".to_string(), "http://127.0.0.1:49173/webhook/rustfs".to_string())]);
|
||||
|
||||
let err = validate_webhook_request(&config)
|
||||
.await
|
||||
.expect_err("a loopback webhook must require an explicit outbound allowlist");
|
||||
let message = err
|
||||
.message()
|
||||
.expect("the validation error should explain the operator action");
|
||||
|
||||
assert!(message.contains("RUSTFS_OUTBOUND_ALLOW_ORIGINS=http://127.0.0.1:49173"));
|
||||
assert!(message.contains("RustFS restarted"), "unexpected message: {message}");
|
||||
assert!(!message.contains("/webhook/rustfs"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn public_webhook_still_passes_outbound_preflight() {
|
||||
let config = HashMap::from([("endpoint".to_string(), "https://hooks.example/webhook".to_string())]);
|
||||
|
||||
validate_webhook_request(&config)
|
||||
.await
|
||||
.expect("a public HTTPS webhook should pass outbound preflight");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -580,8 +580,6 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo
|
||||
));
|
||||
}
|
||||
|
||||
validate_replication_config_capabilities(config)?;
|
||||
|
||||
let targets = metadata_sys::get_bucket_targets_config(bucket)
|
||||
.await
|
||||
.map_err(|err| match err {
|
||||
@@ -2379,6 +2377,8 @@ impl DefaultBucketUsecase {
|
||||
} = req.input;
|
||||
info!(bucket = %bucket, "updating bucket replication config");
|
||||
|
||||
validate_replication_config_capabilities(&replication_configuration)?;
|
||||
|
||||
let Some(store) = self.object_store() else {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
||||
};
|
||||
@@ -3106,10 +3106,7 @@ mod tests {
|
||||
#[test]
|
||||
fn validate_replication_config_capabilities_names_unsupported_field() {
|
||||
let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
|
||||
let destination_key_id = "arn:aws:kms:us-east-1:123456789012:key/opaque-key-id";
|
||||
rule.destination.encryption_configuration = Some(s3s::dto::EncryptionConfiguration {
|
||||
replica_kms_key_id: Some(destination_key_id.to_string()),
|
||||
});
|
||||
rule.destination.encryption_configuration = Some(s3s::dto::EncryptionConfiguration::default());
|
||||
let config = ReplicationConfiguration {
|
||||
role: String::new(),
|
||||
rules: vec![rule],
|
||||
@@ -3123,7 +3120,6 @@ mod tests {
|
||||
err.to_string()
|
||||
.contains("Destination.EncryptionConfiguration is not supported")
|
||||
);
|
||||
assert!(!err.to_string().contains(destination_key_id));
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -4395,6 +4391,31 @@ mod tests {
|
||||
assert_eq!(err.code(), &S3ErrorCode::InternalError);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_put_bucket_replication_rejects_unsupported_fields_before_store_or_metadata_write() {
|
||||
let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
|
||||
rule.destination.account = Some("123456789012".to_string());
|
||||
let input = PutBucketReplicationInput::builder()
|
||||
.bucket("test-bucket".to_string())
|
||||
.replication_configuration(ReplicationConfiguration {
|
||||
role: String::new(),
|
||||
rules: vec![rule],
|
||||
})
|
||||
.build()
|
||||
.unwrap();
|
||||
|
||||
let err = DefaultBucketUsecase::without_context()
|
||||
.execute_put_bucket_replication(build_request(input, Method::PUT))
|
||||
.await
|
||||
.expect_err("unsupported fields must be rejected before store access");
|
||||
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(
|
||||
err.message(),
|
||||
Some("replication field Destination.Account is not supported by this RustFS version")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_put_bucket_encryption_returns_internal_error_when_store_uninitialized() {
|
||||
let input = PutBucketEncryptionInput::builder()
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# Guard: outward README and CHANGELOG material must not make an unsupported
|
||||
# FIPS validation or certification claim. The detailed policy and permitted
|
||||
# qualifiers live in docs/operations/kms-cryptographic-compliance.md; this
|
||||
# check intentionally scans only the two public project-facing documents.
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT_DIR="${CHECK_FIPS_WORDING_ROOT:-$(cd "${SCRIPT_DIR}/.." && pwd)}"
|
||||
|
||||
cd "$ROOT_DIR"
|
||||
|
||||
TARGETS=(README.md CHANGELOG.md)
|
||||
FORBIDDEN_PATTERNS=(
|
||||
'FIPS[[:space:]]+(140-[23](/[[:space:]]*140-[23])?[[:space:]]+)?(validated|certified|compliant)'
|
||||
'FIPS[[:space:]]+mode'
|
||||
'FIPS-enabled'
|
||||
'NIST[[:space:]]+(certified|approved)'
|
||||
'CMVP[[:space:]]+certificate'
|
||||
'(meets|satisfies)[[:space:]]+FIPS'
|
||||
)
|
||||
|
||||
status=0
|
||||
|
||||
for target in "${TARGETS[@]}"; do
|
||||
if [[ ! -f "$target" ]]; then
|
||||
printf 'FIPS wording guard failed: %s is missing\n' "$target" >&2
|
||||
status=1
|
||||
continue
|
||||
fi
|
||||
|
||||
for pattern in "${FORBIDDEN_PATTERNS[@]}"; do
|
||||
matches="$(grep -E -i -n -- "$pattern" "$target" || true)"
|
||||
if [[ -n "$matches" ]]; then
|
||||
printf 'FIPS wording guard failed: forbidden pattern /%s/ in %s:\n%s\n' \
|
||||
"$pattern" "$target" "$matches" >&2
|
||||
status=1
|
||||
fi
|
||||
done
|
||||
done
|
||||
|
||||
if [[ "$status" -ne 0 ]]; then
|
||||
printf 'Remove unsupported FIPS validation wording from README.md or CHANGELOG.md.\n' >&2
|
||||
exit "$status"
|
||||
fi
|
||||
|
||||
printf 'FIPS wording guard passed (README.md and CHANGELOG.md contain no forbidden claims).\n'
|
||||
Reference in New Issue
Block a user