Compare commits

...

32 Commits

Author SHA1 Message Date
overtrue b7fe40f4b3 feat(capture): support complete native log range bounds 2026-10-11 12:28:23 +08:00
overtrue 5b872b8e56 feat(capture): apply native entries atomically with durable replay 2026-10-11 11:50:32 +08:00
overtrue ca5be1ef95 feat(capture): persist native Raft entries in explicitly enabled stores 2026-10-11 05:49:17 +08:00
overtrue 9373e8fa3e Bind capture commands and responses to native Raft entries 2026-10-11 03:45:17 +08:00
overtrue cb32965522 feat(capture): preserve exact raft values and membership codec 2026-10-11 02:04:47 +08:00
overtrue fb6cf49a67 feat(capture): validate peer metadata and CNOD codec 2026-10-11 01:25:21 +08:00
overtrue 8c97049e68 build(capture): pin OpenRaft and guard resolved features 2026-10-11 00:56:30 +08:00
overtrue 17d4fe27cc feat(capture): apply committed Created decisions atomically 2026-10-10 23:35:52 +08:00
overtrue fb0c7972b2 feat(capture): persist committed log boundary 2026-10-10 23:10:28 +08:00
overtrue eba7967755 feat(capture): persist atomic log suffix truncation 2026-10-10 22:59:26 +08:00
overtrue 451c1c31ad feat(capture): add durable local authority storage 2026-10-10 22:33:44 +08:00
overtrue 8780dd670e test(ecstore): verify publication across process kills 2026-10-10 20:19:46 +08:00
overtrue f29bf2c902 fix(iam): preserve empty identity read failures 2026-10-10 11:37:07 +08:00
overtrue fd03436090 feat(ecstore): expose unbounded preserve-empty config reads 2026-10-10 11:10:27 +08:00
overtrue 1f1e7b1c97 fix(iam): preserve identity decryption errors 2026-10-10 10:37:15 +08:00
overtrue 45a952609e test(iam): add real identity storage fixtures 2026-10-10 10:01:03 +08:00
overtrue 323190a95a test(test-utils): support caller-owned instance contexts 2026-10-10 09:43:01 +08:00
overtrue 38cdda9292 fix(ecstore): expose config reader without private bounds 2026-10-10 09:11:47 +08:00
overtrue 85c84fa73d feat(policy): add dormant principal incarnation records 2026-10-10 08:12:14 +08:00
overtrue 51402087c5 feat(ecstore): expose bounded config snapshot reader 2026-10-10 08:03:27 +08:00
overtrue a994984fe2 build(policy): prepare identity incarnation UUID dependency 2026-10-10 06:57:51 +08:00
overtrue 48eb0e402c test(iam): use identity constructor in existing fixtures 2026-10-10 06:52:04 +08:00
GatewayJ 446479791e refactor(iam): remove legacy OIDC federation paths (#8433)
Co-authored-by: Hauser <housemecn@gmail.com>
2026-10-09 07:11:50 -07:00
Chris 99af59a914 Merge commit from fork 2026-10-09 21:43:53 +08:00
Chris b77bd26874 Merge commit from fork 2026-10-09 21:43:08 +08:00
Chris 956c1cf027 Merge commit from fork 2026-10-09 21:40:59 +08:00
Chris 8b9d44110e Merge commit from fork 2026-10-09 21:40:08 +08:00
Chris 44ce8d114f Merge commit from fork 2026-10-09 21:39:18 +08:00
Chris 769b0d7578 Merge commit from fork 2026-10-09 21:28:19 +08:00
Chris 03e775947d Merge commit from fork 2026-10-09 21:18:27 +08:00
Hauser 25e73d3bd6 fix(zip): statically link liblzma on macOS (#8434)
* fix(zip): statically link liblzma on macOS

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(zip): pin macOS liblzma minimum to 0.4.8

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(admin): route bucket recovery errors through facade

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): group version pagination markers

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): avoid cloning copy purge identity

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-09 16:25:24 +08:00
Hauser a464b45367 fix(rustfs): keep recovery handler within S3 gateway (#8429)
* fix(rustfs): route recovery errors through S3 gateway

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): satisfy strict Clippy pagination checks

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): avoid cloning Copy purge identity

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test: restore one-second drive request budgets

* fix(ecstore): resume pending lock bucket creation

Allow retries to reuse an uncommitted Object Lock bucket intent when the same incarnation remains unretired and all physical volumes are absent. Isolate the recovery entrypoint test from the shared production leader lock.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test: provision stack for deep catalog test futures

Apply the existing 4 MiB test-thread stack setup to table-catalog handler/store tests and the relocated-pool GET regression in both local and CI nextest profiles. These cases overflowed libtest's 2 MiB default stack on Linux.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-10-09 14:08:12 +08:00
72 changed files with 8320 additions and 990 deletions
+10
View File
@@ -92,6 +92,12 @@ setup = 'ecstore-large-stack'
filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
setup = 'ecstore-base-stack'
# These handler/store fixtures and the relocated-pool GET compose deep storage
# futures in libtest threads; on Linux they overflow the default 2 MiB stack.
[[profile.default.scripts]]
filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
setup = 'ecstore-base-stack'
[[profile.default.scripts]]
filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
setup = 'lifecycle-large-stack'
@@ -337,6 +343,10 @@ setup = 'ecstore-large-stack'
filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
setup = 'ecstore-base-stack'
[[profile.ci.scripts]]
filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
setup = 'ecstore-base-stack'
[[profile.ci.scripts]]
filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
setup = 'lifecycle-large-stack'
+11
View File
@@ -440,6 +440,17 @@ jobs:
fi
done
- name: Verify macOS binary has no dynamic liblzma dependency
if: matrix.platform == 'macos'
shell: bash
run: |
binary="target/${{ matrix.target }}/release/rustfs"
if otool -L "$binary" | grep -E 'liblzma.*\.dylib'; then
echo "macOS release binary must not depend on a dynamic liblzma library" >&2
exit 1
fi
"$binary" --version
- name: Generate the CPU symbol catalog
if: endsWith(matrix.target, '-linux-gnu') && matrix.features == 'pyroscope'
shell: bash
Generated
+178 -3
View File
@@ -275,6 +275,15 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "anyerror"
version = "0.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "71add24cc141a1e8326f249b74c41cfd217aeb2a67c9c6cf9134d175469afd49"
dependencies = [
"serde",
]
[[package]]
name = "anyhow"
version = "1.0.104"
@@ -1706,6 +1715,30 @@ dependencies = [
"syn 3.0.6",
]
[[package]]
name = "borsh"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
dependencies = [
"borsh-derive",
"bytes",
"cfg_aliases",
]
[[package]]
name = "borsh-derive"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "12cdfe656708a01f89b451a7d36466e6fe6c414de0aa18fc54f864f6f9ca9f56"
dependencies = [
"once_cell",
"proc-macro-crate",
"proc-macro2",
"quote",
"syn 3.0.6",
]
[[package]]
name = "brotli"
version = "8.0.4"
@@ -1772,6 +1805,18 @@ version = "3.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
[[package]]
name = "byte-unit"
version = "5.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c719d56f7e96194cfc53460976d3ba51c85719747c9c62ed99981847b551152b"
dependencies = [
"rust_decimal",
"schemars 1.2.2",
"serde",
"utf8-width",
]
[[package]]
name = "bytemuck"
version = "1.25.2"
@@ -3724,13 +3769,34 @@ dependencies = [
"syn 2.0.119",
]
[[package]]
name = "derive_more"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
dependencies = [
"derive_more-impl 1.0.0",
]
[[package]]
name = "derive_more"
version = "2.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
dependencies = [
"derive_more-impl",
"derive_more-impl 2.1.1",
]
[[package]]
name = "derive_more-impl"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
"unicode-xid",
]
[[package]]
@@ -6234,7 +6300,7 @@ dependencies = [
"bytes",
"chrono",
"dashmap",
"derive_more",
"derive_more 2.1.1",
"futures-util",
"getrandom 0.4.3",
"lazy_static",
@@ -6388,6 +6454,12 @@ dependencies = [
"quote",
]
[[package]]
name = "maplit"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
[[package]]
name = "matchers"
version = "0.2.0"
@@ -7329,6 +7401,42 @@ dependencies = [
"url",
]
[[package]]
name = "openraft"
version = "0.9.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd"
dependencies = [
"anyerror",
"byte-unit",
"chrono",
"clap",
"derive_more 1.0.0",
"futures",
"maplit",
"openraft-macros",
"rand 0.8.8",
"serde",
"thiserror 1.0.69",
"tokio",
"tracing",
"tracing-futures",
"validit",
]
[[package]]
name = "openraft-macros"
version = "0.9.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "55417d79f46a787dd88c043d831d8b7a0713f7c69d1925f9de51cc42c437250f"
dependencies = [
"chrono",
"proc-macro2",
"quote",
"semver",
"syn 2.0.119",
]
[[package]]
name = "openssl-probe"
version = "0.2.1"
@@ -8958,6 +9066,15 @@ dependencies = [
"syn 2.0.119",
]
[[package]]
name = "redb"
version = "4.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb338a6c67830a61bed824b78c2bb034ab1ff401a20616bd7957524f4fdc2f22"
dependencies = [
"libc",
]
[[package]]
name = "redis"
version = "1.7.1"
@@ -9464,6 +9581,23 @@ dependencies = [
"walkdir",
]
[[package]]
name = "rust_decimal"
version = "1.43.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7653272e75dcac41dc199fbea6f5797633994fafd339943c06c9af16bf29cd3a"
dependencies = [
"arrayvec",
"borsh",
"bytes",
"num-traits",
"rand 0.8.8",
"rand 0.9.5",
"serde",
"serde_json",
"wasm-bindgen",
]
[[package]]
name = "rustc-demangle"
version = "0.1.28"
@@ -9666,6 +9800,19 @@ dependencies = [
"url",
]
[[package]]
name = "rustfs-capture-authority"
version = "1.0.1"
dependencies = [
"openraft",
"redb",
"serde",
"serde_json",
"tempfile",
"thiserror 2.0.21",
"url",
]
[[package]]
name = "rustfs-checksums"
version = "1.0.1"
@@ -10473,6 +10620,7 @@ dependencies = [
"tokio",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
@@ -10594,6 +10742,7 @@ dependencies = [
"faster-hex",
"futures",
"hex-simd",
"hmac 0.13.0",
"hotpath",
"http 1.5.0",
"http-body-util",
@@ -11060,6 +11209,7 @@ dependencies = [
"async-compression",
"futures",
"hotpath",
"liblzma",
"rustfs-rio",
"rustfs-tokio-tar",
"serde",
@@ -13044,6 +13194,16 @@ dependencies = [
"tracing-subscriber",
]
[[package]]
name = "tracing-futures"
version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
dependencies = [
"pin-project",
"tracing",
]
[[package]]
name = "tracing-log"
version = "0.2.0"
@@ -13194,7 +13354,7 @@ checksum = "79acf4b257aa56dd7f00466a81693d6c79c362c55092b3137909d6e76249ad80"
dependencies = [
"async-trait",
"chrono",
"derive_more",
"derive_more 2.1.1",
"libc",
"md-5 0.10.6",
"thiserror 2.0.21",
@@ -13294,6 +13454,12 @@ version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]]
name = "utf8-width"
version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "159a7cadce548703edd50d24069bc294c5415ecab0a480e0cd1ca06d112dc94a"
[[package]]
name = "utf8_iter"
version = "1.0.4"
@@ -13320,6 +13486,15 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "validit"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4417835826b78aa6265724624d424df2b1e6e71a33329227d8b04a88950780dc"
dependencies = [
"anyerror",
]
[[package]]
name = "valuable"
version = "0.1.1"
+3
View File
@@ -18,6 +18,7 @@ members = [
"crates/audit", # Audit target management system with multi-target fan-out
"crates/checksums", # client checksums
"crates/common", # Shared utilities and data structures
"crates/capture-authority", # Durable capture authority storage
"crates/data-usage", # Shared data usage models and algorithms
"crates/config", # Configuration management
"crates/connect-symbol-catalog", # Release-only CPU profile symbol catalog generator
@@ -314,6 +315,8 @@ ratelimit = "2.0.0"
rayon = "1.12.0"
rustfs-erasure-codec = { version = "9.0.0" }
reed-solomon-simd = "3.1.0"
openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }
redb = "=4.3.0"
regex = { version = "1.13.1" }
rumqttc = { package = "rumqttc-next", version = "0.34.0" }
redis = { version = "1.7.1" }
+22
View File
@@ -0,0 +1,22 @@
[package]
name = "rustfs-capture-authority"
version.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
repository.workspace = true
description = "Durable local storage for capture authority"
[dependencies]
openraft = { workspace = true }
redb = { workspace = true }
serde = { workspace = true, features = ["derive"] }
thiserror = { workspace = true }
url = { workspace = true }
[dev-dependencies]
serde_json = { workspace = true }
tempfile = { workspace = true }
[lints]
workspace = true
+862
View File
@@ -0,0 +1,862 @@
fn read_native_range_in<R: std::ops::RangeBounds<u64>>(
tx: &redb::ReadTransaction,
range: R,
) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
use std::ops::Bound::{Excluded, Included, Unbounded};
let logs = tx.open_table(LOGS)?;
let start = match range.start_bound() {
Included(value) => *value,
Excluded(value) => match value.checked_add(1) {
Some(value) => value,
None => return Ok(Vec::new()),
},
Unbounded => 0,
};
let end = match range.end_bound() {
Included(value) => *value,
Excluded(value) => match value.checked_sub(1) {
Some(value) => value,
None => return Ok(Vec::new()),
},
Unbounded => u64::MAX,
};
if start > end {
return Ok(Vec::new());
}
let mut entries = Vec::new();
for row in logs.range(start..=end)? {
let (key, value) = row?;
let entry = decode_log(value.value())?;
if entry.id.index != key.value() {
return Err(StoreError::CorruptRecord);
}
entries.push(decode_native_entry(entry)?);
}
Ok(entries)
}
use crate::{
LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
next_index, number,
};
use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
use std::collections::BTreeMap;
pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
pub(crate) const VERSIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("versions");
pub(crate) const EVENTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("events");
pub(crate) const OPERATIONS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("operations");
pub(crate) const APPLY_RESULTS: TableDefinition<u64, &[u8]> = TableDefinition::new("apply_results");
fn version_key(object: &ObjectIdentity, version: [u8; 16]) -> Result<Vec<u8>, StoreError> {
let mut b = object_bytes(object)?;
b.extend(version);
Ok(b)
}
fn decode_version(b: &[u8]) -> Result<CreatedVersion, StoreError> {
let mut d = Decoder::new(b);
let v = CreatedVersion {
prepared: d.prepared()?,
decision_log_id: d.id()?,
binding_revision: d.u64()?,
};
d.finish()?;
Ok(v)
}
fn decode_event(b: &[u8]) -> Result<CreatedEvent, StoreError> {
let mut d = Decoder::new(b);
let e = CreatedEvent {
object: d.object()?,
prepared: d.prepared()?,
decision_log_id: d.id()?,
binding: d.binding()?,
};
d.finish()?;
Ok(e)
}
fn decode_operation(b: &[u8]) -> Result<CreatedOperation, StoreError> {
let mut d = Decoder::new(b);
let len = usize::try_from(d.u64()?).map_err(|_| StoreError::CorruptRecord)?;
let command = b
.get(8..8usize.checked_add(len).ok_or(StoreError::CorruptRecord)?)
.ok_or(StoreError::CorruptRecord)?;
DecideCreated::decode(command)?;
let mut d = Decoder::new(&b[8 + len..]);
let result = d.result()?;
d.finish()?;
Ok(CreatedOperation {
canonical_command: command.to_vec(),
result,
})
}
fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
let mut d = Decoder::new(b);
let r = CreatedApplyResult {
log_id: d.id()?,
result: d.result()?,
};
d.finish()?;
Ok(r)
}
fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
let mut d = Decoder::new(b.value());
let binding = d.binding()?;
d.finish()?;
Ok(binding)
}
fn read_applied(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<Option<LogId>, StoreError> {
let b = meta.get("last_applied")?.ok_or(StoreError::CorruptRecord)?;
let mut d = Decoder::new(b.value());
let id = match d.take::<1>()?[0] {
0 => None,
1 => Some(d.id()?),
_ => return Err(StoreError::CorruptRecord),
};
d.finish()?;
Ok(id)
}
fn required<T>(value: Option<T>) -> Result<T, StoreError> {
value.ok_or(StoreError::CorruptRecord)
}
fn encode_native_entry(entry: &crate::CaptureRaftEntry) -> Result<Vec<u8>, StoreError> {
let id: LogId = entry.log_id.into();
let (kind, body) = match &entry.payload {
openraft::EntryPayload::Blank => (0, vec![]),
openraft::EntryPayload::Normal(command) => (1, command.created().encode()?),
openraft::EntryPayload::Membership(membership) => {
let stored = openraft::StoredMembership::new(Some(entry.log_id), membership.clone());
let checked = crate::CaptureMembership::try_from(&stored)?;
(2, checked.encode()?)
}
};
let mut payload = b"CENR\x01".to_vec();
payload.push(kind);
payload.extend(
u64::try_from(body.len())
.map_err(|_| StoreError::CorruptRecord)?
.to_be_bytes(),
);
payload.extend(body);
encode_log(&crate::LogEntry { id, payload })
}
pub(crate) fn decode_native_entry(entry: crate::LogEntry) -> Result<crate::CaptureRaftEntry, StoreError> {
let bytes = &entry.payload;
if bytes.len() < 14 || &bytes[..5] != b"CENR\x01" {
return Err(StoreError::CorruptRecord);
}
let length = usize::try_from(number(&bytes[6..14])?).map_err(|_| StoreError::CorruptRecord)?;
let body = &bytes[14..];
if length != body.len() {
return Err(StoreError::CorruptRecord);
}
let payload = match bytes[5] {
0 if body.is_empty() => openraft::EntryPayload::Blank,
1 => openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(DecideCreated::decode(body)?)?),
2 => {
let membership = crate::CaptureMembership::decode(body).map_err(|_| StoreError::CorruptRecord)?;
if *membership.log_id() != Some(entry.id) {
return Err(StoreError::CorruptRecord);
}
let stored = openraft::StoredMembership::try_from(&membership).map_err(|_| StoreError::CorruptRecord)?;
openraft::EntryPayload::Membership(stored.membership().clone())
}
_ => return Err(StoreError::CorruptRecord),
};
Ok(crate::CaptureRaftEntry {
log_id: entry.id.into(),
payload,
})
}
fn apply_created(
tx: &redb::WriteTransaction,
id: LogId,
command: &DecideCreated,
canonical: &[u8],
binding: &CaptureBinding,
) -> Result<CreatedResult, StoreError> {
let mut operations = tx.open_table(OPERATIONS)?;
let existing = operations
.get(command.operation_id.as_slice())?
.map(|v| decode_operation(v.value()))
.transpose()?;
let result = if let Some(existing) = existing {
if existing.canonical_command == canonical {
existing.result
} else {
CreatedResult::OperationConflict
}
} else {
let mut heads = tx.open_table(HEADS)?;
let mut versions = tx.open_table(VERSIONS)?;
let object = object_bytes(&command.object)?;
let key = version_key(&command.object, command.prepared.version_id)?;
let actual = heads
.get(object.as_slice())?
.map(|v| {
let mut d = Decoder::new(v.value());
let id = d.uuid()?;
d.finish()?;
Ok::<_, StoreError>(id)
})
.transpose()?;
let outcome = if command.binding_revision != binding.revision {
CreatedResult::BindingMismatch
} else if actual != command.expected_head {
CreatedResult::HeadMismatch { actual_head: actual }
} else if versions.get(key.as_slice())?.is_some() {
CreatedResult::VersionConflict
} else {
let mut version = Vec::new();
put_prepared(&mut version, &command.prepared);
version.extend(encode_id(id));
version.extend(binding.revision.to_be_bytes());
versions.insert(key.as_slice(), version.as_slice())?;
let mut event = object;
put_prepared(&mut event, &command.prepared);
event.extend(encode_id(id));
event.extend(binding_bytes(&binding)?);
tx.open_table(EVENTS)?
.insert(command.operation_id.as_slice(), event.as_slice())?;
heads.insert(object_bytes(&command.object)?.as_slice(), command.prepared.version_id.as_slice())?;
CreatedResult::Created {
version_id: command.prepared.version_id,
event_id: command.operation_id,
decision_log_id: id,
}
};
let mut op = u64::try_from(canonical.len())
.map_err(|_| StoreError::CorruptRecord)?
.to_be_bytes()
.to_vec();
op.extend(canonical);
op.extend(result_bytes(&outcome));
operations.insert(command.operation_id.as_slice(), op.as_slice())?;
outcome
};
Ok(result)
}
fn validate_response_kind(entry: &crate::CaptureRaftEntry, response: &crate::CaptureResponse) -> Result<(), StoreError> {
let kind = match entry.payload {
openraft::EntryPayload::Blank => crate::CaptureResponseKind::Blank,
openraft::EntryPayload::Normal(_) => crate::CaptureResponseKind::Created,
openraft::EntryPayload::Membership(_) => crate::CaptureResponseKind::Membership,
};
if response.kind() != kind || *response.log_id() != LogId::from(entry.log_id) {
return Err(StoreError::CorruptRecord);
}
Ok(())
}
fn encode_response(response: &crate::CaptureResponse) -> Result<Vec<u8>, StoreError> {
let (kind, body) = match response.kind() {
crate::CaptureResponseKind::Blank => (0, vec![]),
crate::CaptureResponseKind::Created => (1, result_bytes(response.created_result().ok_or(StoreError::CorruptRecord)?)),
crate::CaptureResponseKind::Membership => (2, vec![]),
};
let mut bytes = b"CRSP\x01".to_vec();
bytes.push(kind);
bytes.extend(encode_id(*response.log_id()));
bytes.extend(
u64::try_from(body.len())
.map_err(|_| StoreError::CorruptRecord)?
.to_be_bytes(),
);
bytes.extend(body);
Ok(bytes)
}
fn decode_response(bytes: &[u8]) -> Result<crate::CaptureResponse, StoreError> {
if bytes.len() < 38
|| &bytes[..5] != b"CRSP\x01"
|| number(&bytes[30..38])? != u64::try_from(bytes.len() - 38).map_err(|_| StoreError::CorruptRecord)?
{
return Err(StoreError::CorruptRecord);
}
let id = crate::decode_id(&bytes[6..30])?;
match bytes[5] {
0 if bytes.len() == 38 => crate::CaptureResponse::blank(id),
1 => {
let mut body = encode_id(id);
body.extend(&bytes[38..]);
crate::CaptureResponse::created(decode_apply(&body)?)
}
2 if bytes.len() == 38 => crate::CaptureResponse::membership(id),
_ => Err(StoreError::CorruptRecord),
}
}
fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated, StoreError> {
if native {
match decode_native_entry(entry.clone())?.payload {
openraft::EntryPayload::Normal(command) => Ok(command.created().clone()),
_ => Err(StoreError::CorruptRecord),
}
} else {
DecideCreated::decode(&entry.payload)
}
}
pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
let meta = tx.open_table(META)?;
if number(required(meta.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
let last = read_applied(&meta)?;
let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
.map_err(|_| StoreError::CorruptRecord)?;
if membership
.log_id()
.is_some_and(|id| last.is_none_or(|last| id.index > last.index))
{
return Err(StoreError::CorruptRecord);
}
Ok((last, membership))
}
impl Store {
/// Explicitly opt an empty, bound store into native entry persistence.
pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
let tx = self.write_transaction()?;
{
let mut meta = tx.open_table(META)?;
let schema = number(required(meta.get("schema")?)?.value())?;
if !matches!(schema, 3 | 4) {
return Err(StoreError::UnsupportedSchema);
}
read_binding(&meta)?;
if schema == 4 {
let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
.map_err(|_| StoreError::CorruptRecord)?;
let last = read_applied(&meta)?;
if membership
.log_id()
.is_some_and(|id| last.is_none_or(|last| id.index > last.index))
|| (last.is_none() && (!membership.configs().is_empty() || !membership.nodes().is_empty()))
{
return Err(StoreError::CorruptRecord);
}
return Ok(());
}
if read_applied(&meta)?.is_some() || !tx.open_table(LOGS)?.is_empty()? {
return Err(StoreError::UnsupportedSchema);
}
for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
if !tx.open_table(table)?.is_empty()? {
return Err(StoreError::UnsupportedSchema);
}
}
if !tx.open_table(APPLY_RESULTS)?.is_empty()? || meta.get("committed")?.is_some() || meta.get("tail")?.is_some() {
return Err(StoreError::UnsupportedSchema);
}
if meta.get("native_membership")?.is_some() {
return Err(StoreError::CorruptRecord);
}
let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
meta.insert("native_membership", membership.encode()?.as_slice())?;
meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
}
self.commit(tx)
}
pub fn append_native(&mut self, entries: &[crate::CaptureRaftEntry]) -> Result<(), StoreError> {
let tx = self.write_transaction()?;
{
let mut meta = tx.open_table(META)?;
if number(required(meta.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
if entries.is_empty() {
return Ok(());
}
let tail = meta.get("tail")?.map(|value| crate::decode_id(value.value())).transpose()?;
let mut next = next_index(tail.map(|id| id.index))?;
let mut encoded = Vec::with_capacity(entries.len());
for (position, entry) in entries.iter().enumerate() {
if entry.log_id.index != next {
return Err(StoreError::InvalidLogSequence);
}
encoded.push(encode_native_entry(entry)?);
if position + 1 < entries.len() {
next = next_index(Some(next))?;
}
}
let mut logs = tx.open_table(LOGS)?;
for (entry, bytes) in entries.iter().zip(&encoded) {
logs.insert(entry.log_id.index, bytes.as_slice())?;
}
let last = entries.last().ok_or(StoreError::InvalidLogSequence)?;
meta.insert("tail", encode_id(last.log_id.into()).as_slice())?;
}
self.commit(tx)
}
pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
let tx = self.database.begin_read()?;
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
tx.open_table(LOGS)?
.get(index)?
.map(|value| {
let entry = decode_log(value.value())?;
if entry.id.index != index {
return Err(StoreError::CorruptRecord);
}
decode_native_entry(entry)
})
.transpose()
}
/// Read typed native entries within the requested bounds.
pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
let tx = self.database.begin_read()?;
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
read_native_range_in(&tx, range)
}
pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
let tx = self.database.begin_read()?;
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
if start > end {
return Err(StoreError::InvalidLogSequence);
}
read_native_range_in(&tx, start..end)
}
pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
let tx = self.write_transaction()?;
let response;
{
let mut meta = tx.open_table(META)?;
if number(required(meta.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
let binding = read_binding(&meta)?;
let logs = tx.open_table(LOGS)?;
let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
let entry = decode_log(required(logs.get(id.index)?)?.value())?;
if entry.id != id || id.index > committed.index {
return Err(StoreError::InvalidCommittedLog);
}
let native = decode_native_entry(entry)?;
let last = read_applied(&meta)?;
let mut results = tx.open_table(APPLY_RESULTS)?;
if last.is_some_and(|last| id.index <= last.index) {
let saved = decode_response(required(results.get(id.index)?)?.value())?;
if *saved.log_id() != id {
return Err(StoreError::InvalidApplySequence);
}
validate_response_kind(&native, &saved)?;
return Ok(saved);
}
if id.index != next_index(last.map(|last| last.index))? {
return Err(StoreError::InvalidApplySequence);
}
response = match native.payload {
openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
openraft::EntryPayload::Normal(command) => {
let canonical = command.created().encode()?;
let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
}
openraft::EntryPayload::Membership(membership) => {
let stored = openraft::StoredMembership::new(Some(id.into()), membership);
let checked = crate::CaptureMembership::try_from(&stored)?;
meta.insert("native_membership", checked.encode()?.as_slice())?;
crate::CaptureResponse::membership(id)?
}
};
results.insert(id.index, encode_response(&response)?.as_slice())?;
let mut last = vec![1];
last.extend(encode_id(id));
meta.insert("last_applied", last.as_slice())?;
}
self.commit(tx)?;
Ok(response)
}
pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
let tx = self.database.begin_read()?;
if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
return Err(StoreError::UnsupportedSchema);
}
let results = tx.open_table(APPLY_RESULTS)?;
results.get(index)?.map(|v| decode_response(v.value())).transpose()
}
pub fn native_application_state(&self) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
native_application_state(&self.database.begin_read()?)
}
/// Bootstrap the immutable capture policy. The caller must be the trusted
/// deployment provisioning owner. This crate does not authenticate callers;
/// do not expose this public method through an ordinary object request API.
/// Prepared identities assert identity only, not physical data durability.
pub fn pin_capture_binding(&mut self, binding: CaptureBinding) -> Result<(), StoreError> {
let encoded = binding_bytes(&binding)?;
let tx = self.write_transaction()?;
{
let mut meta = tx.open_table(META)?;
if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
if read_binding(&meta)? == binding {
return Ok(());
}
return Err(StoreError::BindingConflict);
}
if meta.get("capture_binding")?.is_some() || meta.get("last_applied")?.is_some() {
return Err(StoreError::CorruptRecord);
}
for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
if !tx.open_table(table)?.is_empty()? {
return Err(StoreError::CorruptRecord);
}
}
if !tx.open_table(APPLY_RESULTS)?.is_empty()? {
return Err(StoreError::CorruptRecord);
}
meta.insert("capture_binding", encoded.as_slice())?;
meta.insert("last_applied", &[0][..])?;
meta.insert("schema", 3u64.to_be_bytes().as_slice())?;
}
self.commit(tx)
}
/// Apply the next exact persisted committed Created command atomically.
pub fn apply_next_created(&mut self, id: LogId) -> Result<CreatedResult, StoreError> {
let tx = self.write_transaction()?;
let result;
{
let mut meta = tx.open_table(META)?;
if number(required(meta.get("schema")?)?.value())? != 3 {
return Err(StoreError::UnsupportedSchema);
}
let binding = read_binding(&meta)?;
let logs = tx.open_table(LOGS)?;
let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
let entry = decode_log(required(logs.get(id.index)?)?.value())?;
if entry.id != id || id.index > committed.index {
return Err(StoreError::InvalidCommittedLog);
}
let last = read_applied(&meta)?;
let mut results = tx.open_table(APPLY_RESULTS)?;
if last.is_some_and(|last| id.index <= last.index) {
let saved = decode_apply(required(results.get(id.index)?)?.value())?;
if saved.log_id != id {
return Err(StoreError::InvalidApplySequence);
}
return Ok(saved.result);
}
if id.index != next_index(last.map(|last| last.index))? {
return Err(StoreError::InvalidApplySequence);
}
let command = DecideCreated::decode(&entry.payload)?;
result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
let mut applied = encode_id(id);
applied.extend(result_bytes(&result));
results.insert(id.index, applied.as_slice())?;
let mut last = vec![1];
last.extend(encode_id(id));
meta.insert("last_applied", last.as_slice())?;
}
self.commit(tx)?;
Ok(result)
}
pub fn capture_binding(&self) -> Result<Option<CaptureBinding>, StoreError> {
let tx = self.database.begin_read()?;
let meta = tx.open_table(META)?;
if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
return Ok(None);
}
Ok(Some(read_binding(&meta)?))
}
pub fn last_applied(&self) -> Result<Option<LogId>, StoreError> {
let tx = self.database.begin_read()?;
let meta = tx.open_table(META)?;
if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
return Ok(None);
}
read_applied(&meta)
}
pub fn created_head(&self, object: &ObjectIdentity) -> Result<Option<[u8; 16]>, StoreError> {
let tx = self.database.begin_read()?;
let t = tx.open_table(HEADS)?;
t.get(object_bytes(object)?.as_slice())?
.map(|v| {
let mut d = Decoder::new(v.value());
let id = d.uuid()?;
d.finish()?;
Ok(id)
})
.transpose()
}
pub fn created_version(&self, object: &ObjectIdentity, version: [u8; 16]) -> Result<Option<CreatedVersion>, StoreError> {
let tx = self.database.begin_read()?;
let t = tx.open_table(VERSIONS)?;
t.get(version_key(object, version)?.as_slice())?
.map(|v| decode_version(v.value()))
.transpose()
}
pub fn created_event(&self, operation: [u8; 16]) -> Result<Option<CreatedEvent>, StoreError> {
let tx = self.database.begin_read()?;
let t = tx.open_table(EVENTS)?;
t.get(operation.as_slice())?.map(|v| decode_event(v.value())).transpose()
}
pub fn created_operation(&self, operation: [u8; 16]) -> Result<Option<CreatedOperation>, StoreError> {
let tx = self.database.begin_read()?;
let t = tx.open_table(OPERATIONS)?;
t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).transpose()
}
pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
let tx = self.database.begin_read()?;
let t = tx.open_table(APPLY_RESULTS)?;
if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
t.get(index)?
.map(|v| {
decode_response(v.value()).map(|r| {
r.created_result().cloned().map(|result| CreatedApplyResult {
log_id: *r.log_id(),
result,
})
})
})
.transpose()
.map(Option::flatten)
} else {
t.get(index)?.map(|v| decode_apply(v.value())).transpose()
}
}
}
pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
let meta = tx.open_table(META)?;
let schema = number(required(meta.get("schema")?)?.value())?;
let native = schema == 4;
if !matches!(schema, 3 | 4) {
return Ok(());
}
let binding = read_binding(&meta).map_err(|error| match error {
StoreError::BindingNotPinned => StoreError::CorruptRecord,
other => other,
})?;
let last = read_applied(&meta)?;
let logs = tx.open_table(LOGS)?;
let committed = validate_committed(&meta, &logs)?;
let heads = tx.open_table(HEADS)?;
let versions = tx.open_table(VERSIONS)?;
let events = tx.open_table(EVENTS)?;
let operations = tx.open_table(OPERATIONS)?;
let results = tx.open_table(APPLY_RESULTS)?;
if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
return Err(StoreError::CorruptRecord);
}
let mut next = Some(0);
let mut observed = None;
let mut first_operations = BTreeMap::new();
let mut expected_membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
let mut replay_operations = BTreeMap::<[u8; 16], (Vec<u8>, CreatedResult)>::new();
let mut replay_heads = BTreeMap::<Vec<u8>, [u8; 16]>::new();
let mut replay_versions = std::collections::BTreeSet::new();
for row in results.iter()? {
let (k, v) = row?;
let entry = decode_log(required(logs.get(k.value())?)?.value())?;
let r = if native {
let response = decode_response(v.value())?;
let native_entry = decode_native_entry(entry.clone())?;
validate_response_kind(&native_entry, &response)?;
if next != Some(k.value()) || response.log_id().index != k.value() {
return Err(StoreError::CorruptRecord);
}
observed = Some(*response.log_id());
next = k.value().checked_add(1);
match native_entry.payload {
openraft::EntryPayload::Membership(membership) => {
expected_membership =
crate::CaptureMembership::try_from(&openraft::StoredMembership::new(Some(entry.id.into()), membership))?;
continue;
}
openraft::EntryPayload::Blank => continue,
openraft::EntryPayload::Normal(_) => CreatedApplyResult {
log_id: *response.log_id(),
result: response.created_result().cloned().ok_or(StoreError::CorruptRecord)?,
},
}
} else {
decode_apply(v.value())?
};
if !native && (next != Some(k.value()) || r.log_id.index != k.value()) {
return Err(StoreError::CorruptRecord);
}
if entry.id != r.log_id {
return Err(StoreError::CorruptRecord);
}
let c = entry_command(&entry, native)?;
let canonical = c.encode()?;
let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
let canonical_matches = op.canonical_command == canonical;
let expected = if canonical_matches {
op.result
} else {
CreatedResult::OperationConflict
};
if native {
let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
if *previous == canonical {
result.clone()
} else {
CreatedResult::OperationConflict
}
} else {
let object = object_bytes(&c.object)?;
let key = version_key(&c.object, c.prepared.version_id)?;
let actual = replay_heads.get(&object).copied();
let result = if c.binding_revision != binding.revision {
CreatedResult::BindingMismatch
} else if actual != c.expected_head {
CreatedResult::HeadMismatch { actual_head: actual }
} else if replay_versions.contains(&key) {
CreatedResult::VersionConflict
} else {
replay_heads.insert(object, c.prepared.version_id);
replay_versions.insert(key);
CreatedResult::Created {
version_id: c.prepared.version_id,
event_id: c.operation_id,
decision_log_id: r.log_id,
}
};
replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
result
};
if replay != r.result {
return Err(StoreError::CorruptRecord);
}
}
if expected != r.result {
return Err(StoreError::CorruptRecord);
}
if canonical_matches {
first_operations.entry(c.operation_id).or_insert(r.log_id);
}
observed = Some(r.log_id);
next = k.value().checked_add(1)
}
if observed != last {
return Err(StoreError::CorruptRecord);
}
if native {
if native_application_state(tx)?.1 != expected_membership
|| heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
|| versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
{
return Err(StoreError::CorruptRecord);
}
for (object, id) in replay_heads {
if required(heads.get(object.as_slice())?)?.value() != id {
return Err(StoreError::CorruptRecord);
}
}
}
for row in operations.iter()? {
let (k, v) = row?;
let op = decode_operation(v.value())?;
let c = DecideCreated::decode(&op.canonical_command)?;
if k.value() != c.operation_id {
return Err(StoreError::CorruptRecord);
}
let first = required(first_operations.get(&c.operation_id))?;
if let CreatedResult::Created {
version_id,
event_id,
decision_log_id,
} = op.result
{
if version_id != c.prepared.version_id
|| event_id != c.operation_id
|| c.binding_revision != binding.revision
|| decision_log_id != *first
{
return Err(StoreError::CorruptRecord);
}
let event = decode_event(required(events.get(event_id.as_slice())?)?.value())?;
if event
!= (CreatedEvent {
object: c.object.clone(),
prepared: c.prepared.clone(),
decision_log_id,
binding: binding.clone(),
})
{
return Err(StoreError::CorruptRecord);
}
} else if events.get(c.operation_id.as_slice())?.is_some() {
return Err(StoreError::CorruptRecord);
}
}
for row in events.iter()? {
let (k, v) = row?;
let e = decode_event(v.value())?;
if e.binding != binding {
return Err(StoreError::CorruptRecord);
}
let op = decode_operation(required(operations.get(k.value())?)?.value())?;
let c = DecideCreated::decode(&op.canonical_command)?;
if c.object != e.object
|| c.prepared != e.prepared
|| op.result
!= (CreatedResult::Created {
version_id: e.prepared.version_id,
event_id: c.operation_id,
decision_log_id: e.decision_log_id,
})
{
return Err(StoreError::CorruptRecord);
}
let version =
decode_version(required(versions.get(version_key(&e.object, e.prepared.version_id)?.as_slice())?)?.value())?;
if version
!= (CreatedVersion {
prepared: e.prepared.clone(),
decision_log_id: e.decision_log_id,
binding_revision: binding.revision,
})
{
return Err(StoreError::CorruptRecord);
}
}
for row in versions.iter()? {
let (k, v) = row?;
let version = decode_version(v.value())?;
let mut d = Decoder::new(k.value());
let object = d.object()?;
let id = d.uuid()?;
d.finish()?;
if id != version.prepared.version_id || version.binding_revision != binding.revision {
return Err(StoreError::CorruptRecord);
}
let entry = decode_log(required(logs.get(version.decision_log_id.index)?)?.value())?;
let c = entry_command(&entry, native)?;
if entry.id != version.decision_log_id
|| c.object != object
|| c.prepared != version.prepared
|| events.get(c.operation_id.as_slice())?.is_none()
{
return Err(StoreError::CorruptRecord);
}
let head = required(heads.get(object_bytes(&object)?.as_slice())?)?;
let mut d = Decoder::new(head.value());
let head_id = d.uuid()?;
d.finish()?;
let current = decode_version(required(versions.get(version_key(&object, head_id)?.as_slice())?)?.value())?;
if current.decision_log_id.index < version.decision_log_id.index {
return Err(StoreError::CorruptRecord);
}
}
for row in heads.iter()? {
let (k, v) = row?;
let mut d = Decoder::new(k.value());
let object = d.object()?;
d.finish()?;
let mut d = Decoder::new(v.value());
let id = d.uuid()?;
d.finish()?;
decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
}
Ok(())
}
@@ -0,0 +1,531 @@
use super::*;
use crate::application::{APPLY_RESULTS, EVENTS, OPERATIONS, VERSIONS};
use crate::tests::CrashBackend;
use redb::ReadableTableMetadata;
use std::{
io::{BufRead, BufReader, Write},
process::{Command, Stdio},
sync::mpsc,
time::Duration,
};
fn identity() -> NodeIdentity {
NodeIdentity {
cluster_id: [7; 16],
node_id: 0,
}
}
fn binding() -> CaptureBinding {
CaptureBinding {
revision: 9,
binding_id: [8; 16],
target_id: [9; 16],
}
}
fn command() -> DecideCreated {
DecideCreated {
operation_id: [1; 16],
object: ObjectIdentity {
bucket_incarnation: [2; 16],
key: vec![0, 255, 3],
},
expected_head: None,
prepared: PreparedIdentity {
version_id: [3; 16],
preparation_id: [4; 16],
content_digest: [5; 32],
content_length: 123,
},
binding_revision: 9,
}
}
fn id(index: u64) -> LogId {
LogId {
term: 8,
leader_node: 0,
index,
}
}
fn append(store: &mut Store, index: u64, c: &DecideCreated) {
store
.append(&[LogEntry {
id: id(index),
payload: c.encode().expect("encode command"),
}])
.expect("append command");
}
fn apply(store: &mut Store, index: u64, c: &DecideCreated) -> CreatedResult {
append(store, index, c);
store.save_committed(id(index)).expect("commit log");
store.apply_next_created(id(index)).expect("apply command")
}
fn success(c: &DecideCreated, index: u64) -> CreatedResult {
CreatedResult::Created {
version_id: c.prepared.version_id,
event_id: c.operation_id,
decision_log_id: id(index),
}
}
fn complete(store: &Store) {
let c = command();
assert_eq!(store.capture_binding().expect("binding"), Some(binding()));
assert_eq!(
store.created_head(&c.object).expect("head"),
Some(c.prepared.version_id),
"U11 acknowledged head lost"
);
assert_eq!(
store.created_version(&c.object, c.prepared.version_id).expect("version"),
Some(CreatedVersion {
prepared: c.prepared.clone(),
decision_log_id: id(0),
binding_revision: 9
})
);
assert_eq!(
store.created_event(c.operation_id).expect("event"),
Some(CreatedEvent {
object: c.object.clone(),
prepared: c.prepared.clone(),
decision_log_id: id(0),
binding: binding()
}),
"U11 complete event missing"
);
assert_eq!(
store.created_operation(c.operation_id).expect("operation"),
Some(CreatedOperation {
canonical_command: c.encode().expect("encode"),
result: success(&c, 0)
})
);
assert_eq!(
store.created_apply_result(0).expect("result"),
Some(CreatedApplyResult {
log_id: id(0),
result: success(&c, 0)
})
);
assert_eq!(store.last_applied().expect("last applied"), Some(id(0)));
}
fn fixture() -> (tempfile::TempDir, Store) {
let dir = tempfile::tempdir().expect("tempdir");
let mut store = Store::initialize(dir.path().join("db"), identity()).expect("initialize");
store.pin_capture_binding(binding()).expect("pin");
(dir, store)
}
fn reopen(path: &std::path::Path) -> Store {
let result = Store::open(path, identity());
assert!(result.is_ok(), "U11 complete reopen: {:?}", result.as_ref().err());
result.expect("asserted successful reopen")
}
#[test]
fn created_codec_is_canonical_and_strict() {
let c = command();
let b = c.encode().expect("encode");
let mut golden = b"CRTD\x01".to_vec();
golden.extend([1; 16]);
golden.extend([2; 16]);
golden.extend(3u32.to_be_bytes());
golden.extend([0, 255, 3]);
golden.push(0);
golden.extend([3; 16]);
golden.extend([4; 16]);
golden.extend([5; 32]);
golden.extend(123u64.to_be_bytes());
golden.extend(9u64.to_be_bytes());
assert_eq!(b, golden);
assert_eq!(DecideCreated::decode(&b).expect("decode"), c);
let mut variants = Vec::new();
let mut v = c.clone();
v.operation_id = [6; 16];
variants.push(v);
let mut v = c.clone();
v.object.bucket_incarnation = [6; 16];
variants.push(v);
let mut v = c.clone();
v.object.key.push(4);
variants.push(v);
let mut v = c.clone();
v.expected_head = Some([6; 16]);
variants.push(v);
let mut v = c.clone();
v.prepared.version_id = [6; 16];
variants.push(v);
let mut v = c.clone();
v.prepared.preparation_id = [6; 16];
variants.push(v);
let mut v = c.clone();
v.prepared.content_digest = [6; 32];
variants.push(v);
let mut v = c.clone();
v.prepared.content_length += 1;
variants.push(v);
let mut v = c.clone();
v.binding_revision += 1;
variants.push(v);
for v in variants {
assert_ne!(v.encode().expect("variant"), b)
}
for n in 0..b.len() {
assert!(DecideCreated::decode(&b[..n]).is_err())
}
let mut bad = b.clone();
bad[4] = 2;
assert!(DecideCreated::decode(&bad).is_err());
let mut bad = b.clone();
bad[44] = 2;
assert!(DecideCreated::decode(&bad).is_err());
let mut bad = b.clone();
bad.push(0);
assert!(DecideCreated::decode(&bad).is_err());
let mut bad = b;
bad[37..41].copy_from_slice(&4097u32.to_be_bytes());
assert!(DecideCreated::decode(&bad).is_err());
let mut bad = c.clone();
bad.operation_id = [0; 16];
assert!(bad.encode().is_err());
bad = c;
bad.object.key = vec![0; MAX_OBJECT_KEY_LENGTH + 1];
assert!(bad.encode().is_err());
}
#[test]
fn created_committed_order_and_identity_are_enforced() {
let (_dir, mut s) = fixture();
let c = command();
append(&mut s, 0, &c);
append(&mut s, 1, &c);
assert!(s.apply_next_created(id(0)).is_err());
assert_eq!(s.last_applied().expect("no apply"), None);
s.save_committed(id(1)).expect("committed");
assert!(matches!(s.apply_next_created(id(1)), Err(StoreError::InvalidApplySequence)));
let mut wrong = id(0);
wrong.term += 1;
assert!(s.apply_next_created(wrong).is_err());
wrong = id(0);
wrong.leader_node += 1;
assert!(s.apply_next_created(wrong).is_err());
assert_eq!(s.apply_next_created(id(0)).expect("first"), success(&c, 0));
complete(&s);
assert_eq!(s.apply_next_created(id(0)).expect("retry"), success(&c, 0));
complete(&s);
assert!(s.apply_next_created(wrong).is_err());
assert_eq!(s.apply_next_created(id(1)).expect("duplicate log"), success(&c, 0));
assert_eq!(s.last_applied().expect("last"), Some(id(1)));
assert_eq!(
s.created_apply_result(1).expect("result"),
Some(CreatedApplyResult {
log_id: id(1),
result: success(&c, 0)
})
);
let tx = s.database.begin_read().expect("read");
assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 1);
}
#[test]
fn created_head_version_event_and_dedup_reopen_together() {
let (dir, mut s) = fixture();
let c = command();
assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
drop(s);
let mut s = reopen(&dir.path().join("db"));
complete(&s);
assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
let mut changed = c.clone();
changed.prepared.content_digest = [6; 32];
assert_eq!(
apply(&mut s, 2, &changed),
CreatedResult::OperationConflict,
"U11 full canonical operation conflict"
);
assert_eq!(
s.created_operation(c.operation_id)
.expect("original")
.expect("operation")
.result,
success(&c, 0)
);
let mut stale = c.clone();
stale.operation_id = [6; 16];
stale.prepared.version_id = [6; 16];
assert_eq!(
apply(&mut s, 3, &stale),
CreatedResult::HeadMismatch {
actual_head: Some(c.prepared.version_id)
}
);
let mut reuse = c.clone();
reuse.operation_id = [7; 16];
reuse.expected_head = Some(c.prepared.version_id);
assert_eq!(apply(&mut s, 4, &reuse), CreatedResult::VersionConflict);
let mut replacement = stale.clone();
replacement.operation_id = [10; 16];
replacement.expected_head = Some(c.prepared.version_id);
assert_eq!(apply(&mut s, 5, &replacement), success(&replacement, 5));
assert_eq!(s.created_head(&c.object).expect("head"), Some(replacement.prepared.version_id));
assert_eq!(
s.created_version(&c.object, c.prepared.version_id)
.expect("old version")
.expect("version")
.decision_log_id,
id(0)
);
assert!(s.created_event(c.operation_id).expect("old event").is_some());
assert!(s.created_event(replacement.operation_id).expect("new event").is_some());
assert!(s.created_event(stale.operation_id).expect("rejection event").is_none());
assert_eq!(
apply(&mut s, 6, &stale),
CreatedResult::HeadMismatch {
actual_head: Some(c.prepared.version_id)
}
);
drop(s);
let s = reopen(&dir.path().join("db"));
assert_eq!(s.last_applied().expect("last"), Some(id(6)));
let tx = s.database.begin_read().expect("read");
assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 2);
}
#[test]
fn created_binding_is_pinned_and_not_caller_selected() {
let (dir, mut s) = fixture();
let mut bad = command();
bad.binding_revision = 10;
assert_eq!(apply(&mut s, 0, &bad), CreatedResult::BindingMismatch);
assert!(s.created_head(&bad.object).expect("head").is_none());
assert!(s.created_event(bad.operation_id).expect("event").is_none());
assert_eq!(
s.created_operation(bad.operation_id)
.expect("operation")
.expect("saved")
.result,
CreatedResult::BindingMismatch
);
s.pin_capture_binding(binding()).expect("idempotent pin");
let mut changed = binding();
changed.target_id = [10; 16];
assert!(matches!(s.pin_capture_binding(changed), Err(StoreError::BindingConflict)));
drop(s);
let s = reopen(&dir.path().join("db"));
assert_eq!(s.capture_binding().expect("binding"), Some(binding()));
}
#[test]
fn created_schema_compatibility_and_corruption_are_checked() {
let dir = tempfile::tempdir().expect("dir");
let path = dir.path().join("db");
let s = Store::initialize(&path, identity()).expect("schema1");
assert_eq!(s.capture_binding().expect("unmigrated"), None);
drop(s);
let mut s = reopen(&path);
assert_eq!(s.last_applied().expect("no apply"), None);
append(&mut s, 0, &command());
s.save_committed(id(0)).expect("schema2");
drop(s);
let mut s = reopen(&path);
assert_eq!(s.capture_binding().expect("schema2 unchanged"), None);
s.pin_capture_binding(binding()).expect("explicit upgrade");
drop(s);
let mut s = reopen(&path);
assert_eq!(s.capture_binding().expect("schema3"), Some(binding()));
s.apply_next_created(id(0)).expect("apply");
drop(s);
let db = Database::open(&path).expect("raw db");
let tx = db.begin_write().expect("write corruption");
tx.open_table(EVENTS)
.expect("events")
.remove(command().operation_id.as_slice())
.expect("remove reference");
tx.commit().expect("commit corruption");
drop(db);
let result = Store::open(&path, identity());
assert!(
matches!(result, Err(StoreError::CorruptRecord)),
"U11 missing event rejected before expect"
);
let (dir, mut s) = fixture();
assert_eq!(s.read_committed().expect("empty committed"), None);
append(&mut s, 0, &command());
s.save_committed(id(0)).expect("committed schema3");
assert_eq!(s.capture_binding().expect("no downgrade"), Some(binding()));
drop(s);
let path = dir.path().join("db");
let db = Database::open(&path).expect("raw db");
let tx = db.begin_write().expect("write");
let mut marker = vec![1];
marker.extend(encode_id(id(0)));
tx.open_table(META)
.expect("meta")
.insert("last_applied", marker.as_slice())
.expect("corrupt applied");
tx.commit().expect("commit");
drop(db);
assert!(
matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
"U11 last applied missing result rejected"
);
let (dir, s) = fixture();
drop(s);
let path = dir.path().join("db");
let db = Database::open(&path).expect("raw pinned db");
let tx = db.begin_write().expect("missing binding write");
tx.open_table(META)
.expect("meta")
.remove("capture_binding")
.expect("remove binding");
tx.commit().expect("binding corruption commit");
drop(db);
assert!(
matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
"U11 missing pinned binding rejected"
);
let (dir, mut s) = fixture();
let c = command();
assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
drop(s);
let path = dir.path().join("db");
let db = Database::open(&path).expect("raw duplicate db");
let tx = db.begin_write().expect("retarget decision write");
let mut key = created::object_bytes(&c.object).expect("object key");
key.extend(c.prepared.version_id);
let mut version = Vec::new();
created::put_prepared(&mut version, &c.prepared);
version.extend(encode_id(id(1)));
version.extend(binding().revision.to_be_bytes());
tx.open_table(VERSIONS)
.expect("versions")
.insert(key.as_slice(), version.as_slice())
.expect("retarget version");
let mut event = created::object_bytes(&c.object).expect("event object");
created::put_prepared(&mut event, &c.prepared);
event.extend(encode_id(id(1)));
event.extend(created::binding_bytes(&binding()).expect("binding bytes"));
tx.open_table(EVENTS)
.expect("events")
.insert(c.operation_id.as_slice(), event.as_slice())
.expect("retarget event");
let canonical = c.encode().expect("canonical");
let mut operation = u64::try_from(canonical.len()).expect("length").to_be_bytes().to_vec();
operation.extend(canonical);
operation.extend(created::result_bytes(&success(&c, 1)));
tx.open_table(OPERATIONS)
.expect("operations")
.insert(c.operation_id.as_slice(), operation.as_slice())
.expect("retarget original response");
for index in 0..=1 {
let mut result = encode_id(id(index));
result.extend(created::result_bytes(&success(&c, 1)));
tx.open_table(APPLY_RESULTS)
.expect("results")
.insert(index, result.as_slice())
.expect("retarget duplicate response");
}
tx.commit().expect("coherent retarget commit");
drop(db);
assert!(
matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
"U11 original decision cannot retarget duplicate log"
);
}
#[test]
fn created_process_child() {
if let Some(path) = std::env::var_os("U11_CHILD_DB") {
let mut s = Store::initialize(path, identity()).expect("child init");
s.pin_capture_binding(binding()).expect("child pin");
assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
complete(&s);
println!("U11_ACK");
std::io::stdout().flush().expect("flush ACK");
loop {
std::thread::park()
}
}
let (_dir, mut s) = fixture();
assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
complete(&s);
}
#[test]
fn created_acknowledged_apply_survives_child_kill() {
let dir = tempfile::tempdir().expect("dir");
let path = dir.path().join("db");
let mut child = Command::new(std::env::current_exe().expect("test executable"))
.args(["--exact", "application_tests::created_process_child", "--nocapture"])
.env("U11_CHILD_DB", &path)
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()
.expect("spawn");
let stdout = child.stdout.take().expect("stdout");
let (tx, rx) = mpsc::channel();
let reader = std::thread::spawn(move || {
for line in BufReader::new(stdout).lines() {
let line = line.expect("child line");
if line.ends_with("U11_ACK") {
let _ = tx.send(());
break;
}
}
});
let acknowledged = rx.recv_timeout(Duration::from_secs(20));
child.kill().expect("kill child");
child.wait().expect("wait child");
reader.join().expect("reader cleanup");
assert!(acknowledged.is_ok(), "U11 bounded child ACK missing");
complete(&reopen(&path));
}
#[test]
fn created_power_cut_before_and_after_apply_flush() {
let (dir, mut s) = fixture();
append(&mut s, 0, &command());
s.save_committed(id(0)).expect("committed");
drop(s);
let path = dir.path().join("db");
let backend = CrashBackend::new(&path);
let mut s = Store::from_database(backend.database(), identity()).expect("backend");
assert_eq!(s.apply_next_created(id(0)).expect("apply ACK"), success(&command(), 0));
backend.cut();
drop(s);
complete(&reopen(&path));
for sync in 1..=2 {
for after in [false, true] {
let (dir, mut s) = fixture();
append(&mut s, 0, &command());
s.save_committed(id(0)).expect("committed");
drop(s);
let path = dir.path().join("db");
let backend = CrashBackend::new(&path);
let mut s = Store::from_database(backend.database(), identity()).expect("backend");
let (entered, resume) = backend.pause_sync(sync, after);
let writer = std::thread::spawn(move || {
let result = s.apply_next_created(id(0));
(s, result)
});
entered.wait();
backend.cut();
resume.wait();
let (s, result) = writer.join().expect("writer");
drop(s);
let reopened = Store::open(&path, identity());
assert!(
reopened.is_ok(),
"U11 apply cut must reopen whole old/new state sync={sync} after={after}: {:?}",
reopened.as_ref().err()
);
let s = reopened.expect("asserted reopen");
if s.last_applied().expect("last").is_some() {
complete(&s)
} else {
assert!(result.is_err());
assert_eq!(s.created_head(&command().object).expect("old head"), None);
assert_eq!(
s.created_version(&command().object, command().prepared.version_id)
.expect("old version"),
None
);
assert_eq!(s.created_event(command().operation_id).expect("old event"), None);
assert_eq!(s.created_operation(command().operation_id).expect("old op"), None);
assert_eq!(s.created_apply_result(0).expect("old result"), None)
}
eprintln!("U11 cut boundary sync={sync} after={after}");
}
}
}
+63
View File
@@ -0,0 +1,63 @@
use crate::{LOGS, LogId, META, Store, StoreError, decode_id, decode_log, encode_id, number};
use redb::{ReadableDatabase, ReadableTable};
pub(crate) fn validate_committed(
meta: &impl ReadableTable<&'static str, &'static [u8]>,
logs: &impl ReadableTable<u64, &'static [u8]>,
) -> Result<Option<LogId>, StoreError> {
let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
let marker = meta.get("committed")?;
match (schema, marker) {
(1 | 3 | 4, None) => {
if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
return Err(StoreError::CorruptRecord);
}
Ok(None)
}
(2 | 3 | 4, Some(marker)) => {
let id = decode_id(marker.value())?;
let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
if decode_log(record.value())?.id != id {
return Err(StoreError::CorruptRecord);
}
Ok(Some(id))
}
(1 | 2, _) => Err(StoreError::CorruptRecord),
_ => Err(StoreError::UnsupportedSchema),
}
}
impl Store {
/// Read the exact committed boundary, validating it against the stored log.
pub fn read_committed(&self) -> Result<Option<LogId>, StoreError> {
let tx = self.database.begin_read()?;
validate_committed(&tx.open_table(META)?, &tx.open_table(LOGS)?)
}
/// Persist a monotonic committed boundary supplied by the authority owner.
pub fn save_committed(&mut self, id: LogId) -> Result<(), StoreError> {
let tx = self.write_transaction()?;
{
let mut meta = tx.open_table(META)?;
let logs = tx.open_table(LOGS)?;
let previous = validate_committed(&meta, &logs)?;
let record = logs.get(id.index)?.ok_or(StoreError::InvalidCommittedLog)?;
if decode_log(record.value())?.id != id {
return Err(StoreError::InvalidCommittedLog);
}
if let Some(previous) = previous {
if id.index < previous.index {
return Err(StoreError::CommittedRegression);
}
if id == previous {
return Ok(());
}
}
meta.insert("committed", encode_id(id).as_slice())?;
if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
}
}
self.commit(tx)
}
}
+242
View File
@@ -0,0 +1,242 @@
use crate::{LogId, StoreError, decode_id, encode_id};
pub const MAX_OBJECT_KEY_LENGTH: usize = 4096;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObjectIdentity {
pub bucket_incarnation: [u8; 16],
pub key: Vec<u8>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedIdentity {
pub version_id: [u8; 16],
pub preparation_id: [u8; 16],
pub content_digest: [u8; 32],
pub content_length: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecideCreated {
pub operation_id: [u8; 16],
pub object: ObjectIdentity,
pub expected_head: Option<[u8; 16]>,
pub prepared: PreparedIdentity,
pub binding_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CaptureBinding {
pub revision: u64,
pub binding_id: [u8; 16],
pub target_id: [u8; 16],
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CreatedResult {
Created {
version_id: [u8; 16],
event_id: [u8; 16],
decision_log_id: LogId,
},
HeadMismatch {
actual_head: Option<[u8; 16]>,
},
OperationConflict,
BindingMismatch,
VersionConflict,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedVersion {
pub prepared: PreparedIdentity,
pub decision_log_id: LogId,
pub binding_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedEvent {
pub object: ObjectIdentity,
pub prepared: PreparedIdentity,
pub decision_log_id: LogId,
pub binding: CaptureBinding,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedOperation {
pub canonical_command: Vec<u8>,
pub result: CreatedResult,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatedApplyResult {
pub log_id: LogId,
pub result: CreatedResult,
}
pub(crate) struct Decoder<'a> {
bytes: &'a [u8],
}
impl<'a> Decoder<'a> {
pub(crate) fn new(bytes: &'a [u8]) -> Self {
Self { bytes }
}
pub(crate) fn take<const N: usize>(&mut self) -> Result<[u8; N], StoreError> {
let a = self
.bytes
.get(..N)
.ok_or(StoreError::CorruptRecord)?
.try_into()
.map_err(|_| StoreError::CorruptRecord)?;
self.bytes = &self.bytes[N..];
Ok(a)
}
pub(crate) fn u64(&mut self) -> Result<u64, StoreError> {
Ok(u64::from_be_bytes(self.take()?))
}
pub(crate) fn uuid(&mut self) -> Result<[u8; 16], StoreError> {
let id = self.take()?;
if id == [0; 16] {
return Err(StoreError::CorruptRecord);
}
Ok(id)
}
pub(crate) fn option(&mut self) -> Result<Option<[u8; 16]>, StoreError> {
match self.take::<1>()?[0] {
0 => Ok(None),
1 => Ok(Some(self.uuid()?)),
_ => Err(StoreError::CorruptRecord),
}
}
pub(crate) fn object(&mut self) -> Result<ObjectIdentity, StoreError> {
let bucket_incarnation = self.uuid()?;
let len = usize::try_from(u32::from_be_bytes(self.take()?)).map_err(|_| StoreError::CorruptRecord)?;
if len > MAX_OBJECT_KEY_LENGTH {
return Err(StoreError::CorruptRecord);
}
let key = self.bytes.get(..len).ok_or(StoreError::CorruptRecord)?.to_vec();
self.bytes = &self.bytes[len..];
Ok(ObjectIdentity { bucket_incarnation, key })
}
pub(crate) fn prepared(&mut self) -> Result<PreparedIdentity, StoreError> {
Ok(PreparedIdentity {
version_id: self.uuid()?,
preparation_id: self.uuid()?,
content_digest: self.take()?,
content_length: self.u64()?,
})
}
pub(crate) fn id(&mut self) -> Result<LogId, StoreError> {
decode_id(&self.take::<24>()?)
}
pub(crate) fn binding(&mut self) -> Result<CaptureBinding, StoreError> {
Ok(CaptureBinding {
revision: self.u64()?,
binding_id: self.uuid()?,
target_id: self.uuid()?,
})
}
pub(crate) fn finish(self) -> Result<(), StoreError> {
if self.bytes.is_empty() {
Ok(())
} else {
Err(StoreError::CorruptRecord)
}
}
pub(crate) fn result(&mut self) -> Result<CreatedResult, StoreError> {
match self.take::<1>()?[0] {
0 => Ok(CreatedResult::Created {
version_id: self.uuid()?,
event_id: self.uuid()?,
decision_log_id: self.id()?,
}),
1 => Ok(CreatedResult::HeadMismatch {
actual_head: self.option()?,
}),
2 => Ok(CreatedResult::OperationConflict),
3 => Ok(CreatedResult::BindingMismatch),
4 => Ok(CreatedResult::VersionConflict),
_ => Err(StoreError::CorruptRecord),
}
}
}
pub(crate) fn put_option(b: &mut Vec<u8>, id: Option<[u8; 16]>) {
b.push(u8::from(id.is_some()));
if let Some(id) = id {
b.extend(id)
}
}
pub(crate) fn object_bytes(o: &ObjectIdentity) -> Result<Vec<u8>, StoreError> {
if o.bucket_incarnation == [0; 16] || o.key.len() > MAX_OBJECT_KEY_LENGTH {
return Err(StoreError::CorruptRecord);
}
let capacity = 20usize.checked_add(o.key.len()).ok_or(StoreError::CorruptRecord)?;
let mut b = Vec::with_capacity(capacity);
b.extend(o.bucket_incarnation);
b.extend(
u32::try_from(o.key.len())
.map_err(|_| StoreError::CorruptRecord)?
.to_be_bytes(),
);
b.extend(&o.key);
Ok(b)
}
pub(crate) fn put_prepared(b: &mut Vec<u8>, p: &PreparedIdentity) {
b.extend(p.version_id);
b.extend(p.preparation_id);
b.extend(p.content_digest);
b.extend(p.content_length.to_be_bytes())
}
pub(crate) fn binding_bytes(v: &CaptureBinding) -> Result<Vec<u8>, StoreError> {
let mut b = v.revision.to_be_bytes().to_vec();
b.extend(v.binding_id);
b.extend(v.target_id);
let mut d = Decoder::new(&b);
d.binding()?;
d.finish()?;
Ok(b)
}
pub(crate) fn result_bytes(r: &CreatedResult) -> Vec<u8> {
let mut b = Vec::new();
match r {
CreatedResult::Created {
version_id,
event_id,
decision_log_id,
} => {
b.push(0);
b.extend(version_id);
b.extend(event_id);
b.extend(encode_id(*decision_log_id))
}
CreatedResult::HeadMismatch { actual_head } => {
b.push(1);
put_option(&mut b, *actual_head)
}
CreatedResult::OperationConflict => b.push(2),
CreatedResult::BindingMismatch => b.push(3),
CreatedResult::VersionConflict => b.push(4),
}
b
}
impl DecideCreated {
pub fn encode(&self) -> Result<Vec<u8>, StoreError> {
let object = object_bytes(&self.object)?;
let capacity = object.len().checked_add(134).ok_or(StoreError::CorruptRecord)?;
let mut b = Vec::with_capacity(capacity);
b.extend(b"CRTD\x01");
b.extend(self.operation_id);
b.extend(object);
put_option(&mut b, self.expected_head);
put_prepared(&mut b, &self.prepared);
b.extend(self.binding_revision.to_be_bytes());
Self::decode(&b)?;
Ok(b)
}
pub fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
let mut d = Decoder::new(bytes);
if d.take::<5>()? != *b"CRTD\x01" {
return Err(StoreError::CorruptRecord);
}
let c = Self {
operation_id: d.uuid()?,
object: d.object()?,
expected_head: d.option()?,
prepared: d.prepared()?,
binding_revision: d.u64()?,
};
d.finish()?;
Ok(c)
}
}
File diff suppressed because it is too large Load Diff
+371
View File
@@ -0,0 +1,371 @@
//! Validated capture peer metadata and its independent CNOD v1 codec.
use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct CaptureNode {
rpc_uri: String,
}
#[derive(Debug, thiserror::Error)]
pub enum CaptureNodeError {
#[error("invalid RPC URI")]
InvalidRpcUri,
#[error("invalid RPC URI")]
InvalidRpcUriParse(#[source] url::ParseError),
#[error("RPC URI too long")]
RpcUriTooLong,
#[error("invalid node magic")]
InvalidNodeMagic,
#[error("unsupported node version")]
UnsupportedNodeVersion,
#[error("invalid node length")]
InvalidNodeLength,
#[error("truncated node record")]
TruncatedNodeRecord,
#[error("invalid node UTF-8")]
InvalidNodeUtf8(#[source] std::str::Utf8Error),
#[error("trailing node bytes")]
TrailingNodeBytes,
}
impl CaptureNodeError {
pub fn code(&self) -> &'static str {
match self {
Self::InvalidRpcUri | Self::InvalidRpcUriParse(_) => "invalid_rpc_uri",
Self::RpcUriTooLong => "rpc_uri_too_long",
Self::InvalidNodeMagic => "invalid_node_magic",
Self::UnsupportedNodeVersion => "unsupported_node_version",
Self::InvalidNodeLength => "invalid_node_length",
Self::TruncatedNodeRecord => "truncated_node_record",
Self::InvalidNodeUtf8(_) => "invalid_node_utf8",
Self::TrailingNodeBytes => "trailing_node_bytes",
}
}
}
impl CaptureNode {
pub fn new(rpc_uri: String) -> Result<Self, CaptureNodeError> {
let node = Self { rpc_uri };
node.validate()?;
Ok(node)
}
pub fn rpc_uri(&self) -> &str {
&self.rpc_uri
}
pub fn validate(&self) -> Result<(), CaptureNodeError> {
let uri = &self.rpc_uri;
if uri.len() > 4096 {
return Err(CaptureNodeError::RpcUriTooLong);
}
if uri.is_empty() || uri.chars().any(|c| c.is_whitespace() || c.is_control() || c == '\\') {
return Err(CaptureNodeError::InvalidRpcUri);
}
let rest = uri
.strip_prefix("http://")
.or_else(|| uri.strip_prefix("https://"))
.ok_or(CaptureNodeError::InvalidRpcUri)?;
if uri.contains('?') || uri.contains('#') {
return Err(CaptureNodeError::InvalidRpcUri);
}
let (authority, path) = rest.split_once('/').map_or((rest, ""), |(a, p)| (a, p));
// An empty suffix after the first slash is the only allowed path.
if !path.is_empty() || authority.contains('@') {
return Err(CaptureNodeError::InvalidRpcUri);
}
let (host, port) = if authority.starts_with('[') {
let end = authority.find(']').ok_or(CaptureNodeError::InvalidRpcUri)?;
let port = authority
.get(end + 1..)
.and_then(|suffix| suffix.strip_prefix(':'))
.ok_or(CaptureNodeError::InvalidRpcUri)?;
(&authority[..=end], port)
} else {
authority.split_once(':').ok_or(CaptureNodeError::InvalidRpcUri)?
};
if host.is_empty() || port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) {
return Err(CaptureNodeError::InvalidRpcUri);
}
let port = port.parse::<u16>().map_err(|_| CaptureNodeError::InvalidRpcUri)?;
if port == 0 {
return Err(CaptureNodeError::InvalidRpcUri);
}
let parsed_url = url::Url::parse(uri).map_err(CaptureNodeError::InvalidRpcUriParse)?;
if parsed_url.host().is_none() {
return Err(CaptureNodeError::InvalidRpcUri);
}
Ok(())
}
pub fn encode(&self) -> Result<Vec<u8>, CaptureNodeError> {
self.validate()?;
let length = u32::try_from(self.rpc_uri.len()).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
let mut bytes = b"CNOD\x01".to_vec();
bytes.extend_from_slice(&length.to_be_bytes());
bytes.extend_from_slice(self.rpc_uri.as_bytes());
Ok(bytes)
}
pub fn decode(bytes: &[u8]) -> Result<Self, CaptureNodeError> {
if bytes.len() < 9 {
return Err(CaptureNodeError::TruncatedNodeRecord);
}
if &bytes[..4] != b"CNOD" {
return Err(CaptureNodeError::InvalidNodeMagic);
}
if bytes[4] != 1 {
return Err(CaptureNodeError::UnsupportedNodeVersion);
}
let length = u32::from_be_bytes(bytes[5..9].try_into().map_err(|_| CaptureNodeError::InvalidNodeLength)?);
if !(1..=4096).contains(&length) {
return Err(CaptureNodeError::InvalidNodeLength);
}
let length = usize::try_from(length).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
let end = 9_usize.checked_add(length).ok_or(CaptureNodeError::InvalidNodeLength)?;
if bytes.len() < end {
return Err(CaptureNodeError::TruncatedNodeRecord);
}
if bytes.len() != end {
return Err(CaptureNodeError::TrailingNodeBytes);
}
let uri = std::str::from_utf8(&bytes[9..end]).map_err(CaptureNodeError::InvalidNodeUtf8)?;
Self::new(uri.to_owned())
}
}
impl Serialize for CaptureNode {
fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
self.validate().map_err(serde::ser::Error::custom)?;
let mut state = serializer.serialize_struct("CaptureNode", 1)?;
state.serialize_field("rpc_uri", &self.rpc_uri)?;
state.end()
}
}
impl<'de> Deserialize<'de> for CaptureNode {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Fields {
rpc_uri: String,
}
let fields = Fields::deserialize(deserializer).map_err(|_| serde::de::Error::custom("invalid capture node fields"))?;
Self::new(fields.rpc_uri).map_err(serde::de::Error::custom)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::error::Error;
fn record(payload: &[u8]) -> Vec<u8> {
let mut bytes = b"CNOD\x01".to_vec();
bytes.extend_from_slice(&u32::try_from(payload.len()).expect("test payload length").to_be_bytes());
bytes.extend_from_slice(payload);
bytes
}
fn reject_uri(uri: &str) {
let result = CaptureNode::new(uri.to_owned());
assert!(result.is_err(), "URI should reject: {uri:?}");
assert_eq!(result.expect_err("rejected URI").code(), "invalid_rpc_uri");
}
fn reject_record(bytes: &[u8], code: &str) {
let result = CaptureNode::decode(bytes);
assert!(result.is_err(), "record should reject with {code}");
assert_eq!(result.expect_err("rejected record").code(), code);
}
#[test]
fn actual_openraft_node_bound() {
fn require_node<T: openraft::Node>() {}
require_node::<CaptureNode>();
assert_eq!(CaptureNode::default().rpc_uri(), "");
}
#[test]
fn default_and_empty_rejected_at_every_boundary() {
let node = CaptureNode::default();
assert_eq!(node.rpc_uri(), "");
reject_uri("");
assert_eq!(node.validate().expect_err("empty validation").code(), "invalid_rpc_uri");
assert!(serde_json::to_string(&node).is_err());
assert!(serde_json::from_str::<CaptureNode>(r#"{"rpc_uri":""}"#).is_err());
assert_eq!(node.encode().expect_err("empty encoding").code(), "invalid_rpc_uri");
reject_record(&record(b""), "invalid_node_length");
}
#[test]
fn accepted_uris_preserve_exact_bytes() {
for uri in [
"http://node:80",
"https://node:443",
"http://node:1",
"https://node:65535/",
"http://127.0.0.1:9000",
"https://[::1]:443/",
"http://[2001:db8::1]:1",
"http://NoDe:00080/",
] {
let result = CaptureNode::new(uri.to_owned());
assert!(result.is_ok(), "valid URI construction: {uri}");
let node = result.expect("valid URI");
assert_eq!(node.rpc_uri(), uri, "preserve explicit default port and raw URI");
assert_eq!(node.encode().expect("encode URI"), record(uri.as_bytes()));
assert_eq!(CaptureNode::decode(&record(uri.as_bytes())).expect("decode URI"), node);
}
}
#[test]
fn golden_cnod_bytes() {
let node = CaptureNode {
rpc_uri: "http://node:80".to_owned(),
};
let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
assert_eq!(node.encode().ok().as_deref(), Some(golden.as_slice()), "golden CNOD bytes");
}
#[test]
fn valid_cnod_decode() {
let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
let decoded = CaptureNode::decode(golden);
assert!(decoded.is_ok(), "golden decode succeeds");
assert_eq!(decoded.expect("golden decoded node").rpc_uri(), "http://node:80");
}
#[test]
fn serde_shape_roundtrip_and_strict_fields() {
let node = CaptureNode::new("https://node:443/".to_owned()).expect("serde node");
let json = r#"{"rpc_uri":"https://node:443/"}"#;
assert_eq!(serde_json::to_string(&node).expect("serialize node"), json);
assert_eq!(serde_json::from_str::<CaptureNode>(json).expect("deserialize node"), node);
for json in [
r#"{}"#,
r#"{"rpc_uri":1}"#,
r#"{"rpc_uri":null}"#,
r#"{"rpc_uri":"http://node:1","extra":0}"#,
r#"{"rpc_uri":"http://node:1","rpc_uri":"http://node:2"}"#,
] {
assert!(serde_json::from_str::<CaptureNode>(json).is_err(), "strict fields: {json}");
}
}
#[test]
fn credentials_and_raw_uri_shapes_reject() {
for uri in [
"http://user@node:1",
"http://user:password@node:1",
"http://@node:1",
"HTTP://node:1",
"Https://node:1",
"http://node",
"http://node:",
"http://node:0",
"http://node:-1",
"http://node:+1",
"http://node:65536",
"http://node:999999999999999999999",
"http://[::1:1",
"http://[::1]extra:1",
"http://[invalid]:1",
"http://::1:1",
"http://:1",
"/disk/path",
"file:///disk/path",
"http://node:1/./",
"http://node:1/../",
"http://node:1/%2e/",
"http://node:1/%2E%2e/",
"http://node:1/path",
"http://node:1\\",
" http://node:1",
"http://node:1\n",
"http://no\tde:1",
"http://node:1\u{00a0}",
"http://node:1\u{0085}",
"http://node:1\0",
] {
reject_uri(uri);
}
}
#[test]
fn query_and_fragment_markers_reject() {
for uri in [
"http://node:1?",
"http://node:1?x=y",
"http://node:1/?",
"http://node:1#",
"http://node:1#fragment",
"http://node:1/#",
] {
reject_uri(uri);
}
}
#[test]
fn utf8_byte_length_boundaries() {
for length in [4095, 4096] {
let prefix = "http://node:";
let uri = format!("{prefix}{}1", "0".repeat(length - prefix.len() - 1));
assert_eq!(uri.len(), length);
assert!(url::Url::parse(&uri).is_ok(), "boundary fixture accepted by actual URL parser");
let node = CaptureNode::new(uri.clone()).expect("boundary node");
assert_eq!(node.rpc_uri(), uri);
assert_eq!(
CaptureNode::decode(&node.encode().expect("boundary encoding")).expect("boundary decoding"),
node
);
}
let uri = format!("http://node:{}1", "0".repeat(4097 - "http://node:".len() - 1));
assert_eq!(CaptureNode::new(uri).expect_err("oversize URI").code(), "rpc_uri_too_long");
let unicode = format!("http://{}:1", "é".repeat(2044));
assert!(unicode.chars().count() < 4096);
assert!(unicode.len() > 4096);
assert_eq!(CaptureNode::new(unicode).expect_err("UTF-8 byte limit").code(), "rpc_uri_too_long");
}
#[test]
fn cnod_headers_and_lengths_reject_exact_codes() {
let valid = record(b"http://node:80");
for end in 0..9 {
reject_record(&valid[..end], "truncated_node_record");
}
let mut bad = valid.clone();
bad[0] = b'X';
reject_record(&bad, "invalid_node_magic");
bad = valid.clone();
bad[4] = 2;
reject_record(&bad, "unsupported_node_version");
for length in [0_u32, 4097, u32::MAX] {
bad = b"CNOD\x01".to_vec();
bad.extend_from_slice(&length.to_be_bytes());
reject_record(&bad, "invalid_node_length");
}
for end in 9..valid.len() {
reject_record(&valid[..end], "truncated_node_record");
}
reject_record(&record(&[0xff]), "invalid_node_utf8");
reject_record(&record(&[0xc3, 0x28]), "invalid_node_utf8");
reject_record(&record(b"invalid URI"), "invalid_rpc_uri");
}
#[test]
fn trailing_bytes_reject() {
for tail in [&b"x"[..], &b"xyz"[..]] {
let mut bytes = record(b"http://node:80");
bytes.extend_from_slice(tail);
reject_record(&bytes, "trailing_node_bytes");
}
}
#[test]
fn errors_and_sources_redact_input() {
let sentinel = "http://SECRET_USER:SECRET_PASSWORD@node:1";
let error = CaptureNode::new(sentinel.to_owned()).expect_err("credentials reject");
let mut current: &dyn Error = &error;
loop {
let rendered = format!("{current} {current:?}");
assert!(!rendered.contains("SECRET"));
match current.source() {
Some(source) => current = source,
None => break,
}
}
let json = format!(r#"{{"rpc_uri":"{sentinel}"}}"#);
let error = serde_json::from_str::<CaptureNode>(&json).expect_err("serde credentials reject");
assert!(!format!("{error} {error:?}").contains("SECRET"));
let json = format!(r#"{{"{sentinel}":0}}"#);
let error = serde_json::from_str::<CaptureNode>(&json).expect_err("unknown fields reject");
assert!(!format!("{error} {error:?}").contains("SECRET"));
let error = CaptureNode::new("http://[bad]:1".to_owned()).expect_err("parser rejection");
assert!(error.source().is_some(), "retain URL parser source");
let error = CaptureNode::decode(&record(&[0xff])).expect_err("UTF-8 rejection");
assert!(error.source().is_some(), "retain UTF-8 source");
}
}
File diff suppressed because it is too large Load Diff
+43
View File
@@ -0,0 +1,43 @@
use crate::{LOGS, META, Store, StoreError, decode_id, decode_log, encode_id};
use redb::ReadableTable;
impl Store {
/// Atomically remove every log entry at or after `from`.
pub fn truncate_suffix(&mut self, from: u64) -> Result<(), StoreError> {
let tx = self.write_transaction()?;
{
let mut meta = tx.open_table(META)?;
let logs = tx.open_table(LOGS)?;
if crate::committed::validate_committed(&meta, &logs)?.is_some_and(|id| from <= id.index) {
return Err(StoreError::CommittedLogProtected);
}
drop(logs);
let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
if tail.is_none_or(|id| from > id.index) {
return Ok(());
}
let mut logs = tx.open_table(LOGS)?;
let new_tail = if from == 0 {
None
} else {
let index = from - 1;
let record = logs.get(index)?.ok_or(StoreError::InvalidLogSequence)?;
let id = decode_log(record.value())?.id;
if id.index != index {
return Err(StoreError::CorruptRecord);
}
Some(id)
};
logs.retain(|index, _| index < from)?;
match new_tail {
Some(id) => {
meta.insert("tail", encode_id(id).as_slice())?;
}
None => {
meta.remove("tail")?;
}
}
}
self.commit(tx)
}
}
+18
View File
@@ -371,6 +371,24 @@ pub mod config {
with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
with_server_config_write_lock,
};
/// Read config bytes and metadata without treating an empty payload as absent.
pub async fn read_config_preserve_empty_with_metadata_opts(
api: std::sync::Arc<crate::store::ECStore>,
file: &str,
opts: &crate::object_api::ObjectOptions,
) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
crate::config::com::read_config_preserve_empty_with_metadata_opts(api, file, opts).await
}
pub async fn read_config_limited_preserve_empty_with_metadata_opts(
api: std::sync::Arc<crate::store::ECStore>,
file: &str,
opts: &crate::object_api::ObjectOptions,
max_bytes: usize,
) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
crate::config::com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
}
}
pub mod storageclass {
@@ -2939,6 +2939,20 @@ fn resolve_tier_free_version_recovery_enabled(value: Result<String, env::VarErro
}
fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<()>) -> Option<JoinHandle<()>> {
spawn_tier_free_version_recovery_once_with_options(
api,
started,
TIER_FREE_VERSION_RECOVERY_LEADER_LOCK.to_string(),
tier_free_version_recovery_leader_retry_delay,
)
}
fn spawn_tier_free_version_recovery_once_with_options(
api: Arc<ECStore>,
started: &OnceLock<()>,
leader_lock_name: String,
retry_delay: fn() -> StdDuration,
) -> Option<JoinHandle<()>> {
if !tier_free_version_recovery_enabled() {
warn!(
event = EVENT_LIFECYCLE_WORKER_STATE,
@@ -2956,15 +2970,15 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<(
Some(tokio::spawn(async move {
let cancel_token = api.ctx.background_cancel_token().unwrap_or_default();
let acquire_api = Arc::clone(&api);
let acquire_lock_name = Arc::<str>::from(leader_lock_name);
run_tier_free_version_recovery_coordinator(
cancel_token,
tier_free_version_recovery_leader_retry_delay,
retry_delay,
move || {
let api = Arc::clone(&acquire_api);
let lock_name = Arc::clone(&acquire_lock_name);
async move {
let lock = api
.new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK)
.await?;
let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &lock_name).await?;
match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await {
Ok(guard) => Ok(Some(guard)),
Err(LockError::Timeout { .. }) => Ok(None),
@@ -14193,6 +14207,10 @@ mod tests {
.await;
}
fn fast_tier_free_version_recovery_leader_retry_delay() -> StdDuration {
StdDuration::from_millis(10)
}
#[tokio::test]
#[serial]
async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
@@ -14208,8 +14226,13 @@ mod tests {
seed_recoverable_free_version(&disk_paths, &bucket, object, None, None).await;
let started = OnceLock::new();
let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started)
.expect("production recovery entrypoint should start once");
let recovery = super::spawn_tier_free_version_recovery_once_with_options(
Arc::clone(&ecstore),
&started,
format!("tier-free-version-recovery/test/{}", Uuid::new_v4()),
fast_tier_free_version_recovery_leader_retry_delay,
)
.expect("production recovery entrypoint should start once");
let task = tokio::time::timeout(StdDuration::from_secs(30), async { recovery_rx.lock().await.recv().await })
.await
.expect("production recovery entrypoint should enqueue the seeded item")
+24 -4
View File
@@ -1292,14 +1292,34 @@ pub(crate) async fn get_object_lock_config_state_in(
pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
get_object_lock_config_and_incarnation_from_disk_with_guard_in(ctx, bucket, None).await
}
/// Bucket deletion already owns the namespace/lifecycle fences. Reuse its
/// live guard rather than recursively acquiring the bucket namespace lock.
/// The caller must also hold the metadata transaction read lock.
pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_with_guard_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
guard: Option<&rustfs_lock::NamespaceLockGuard>,
) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();
match bucket_meta_sys
.read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
.await?
{
let authority = match guard {
Some(guard) => {
bucket_meta_sys
.read_authoritative_metadata_from_disk_under_guard(bucket, guard)
.await?
}
None => {
bucket_meta_sys
.read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
.await?
}
};
match authority {
BucketMetadataAuthority::Authoritative(metadata)
if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
{
+76 -8
View File
@@ -495,10 +495,22 @@ pub(crate) async fn read_config_preserve_empty<S>(api: Arc<S>, file: &str) -> Re
where
S: EcstoreObjectIO,
{
let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, None).await?;
let (data, _obj) = read_config_preserve_empty_with_metadata_opts(api, file, &ObjectOptions::default()).await?;
Ok(data)
}
/// Preserve an empty config payload, its metadata, and caller-supplied read options.
pub(crate) async fn read_config_preserve_empty_with_metadata_opts<S>(
api: Arc<S>,
file: &str,
opts: &ObjectOptions,
) -> Result<(Vec<u8>, ObjectInfo)>
where
S: EcstoreObjectIO,
{
read_config_with_metadata_inner(api, file, opts, true, None).await
}
pub async fn read_config_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
where
S: ObjectIO<
@@ -527,15 +539,13 @@ pub(crate) async fn read_config_no_lock_preserve_empty_with_metadata<S>(api: Arc
where
S: EcstoreObjectIO,
{
read_config_with_metadata_inner(
read_config_preserve_empty_with_metadata_opts(
api,
file,
&ObjectOptions {
no_lock: true,
..Default::default()
},
true,
None,
)
.await
}
@@ -2815,9 +2825,10 @@ mod tests {
config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
heal_config_descriptor, is_standard_object_server_config, lookup_configs, map_system_metadata_write_error,
new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty,
read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot, save_config_with_opts_inner,
save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
read_config_preserve_empty_with_metadata_opts, read_config_with_metadata, read_config_without_migrate,
read_server_config_snapshot, save_config_with_opts_inner, save_server_config, save_server_config_snapshot,
save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
storage_class_kvs_mut,
};
use crate::config::{audit, heal, notify, oidc, scanner};
use crate::disk::{RUSTFS_META_BUCKET, endpoint::Endpoint};
@@ -5065,6 +5076,7 @@ mod tests {
heal_replacement: Option<Vec<u8>>,
heal_calls: AtomicUsize,
write_calls: AtomicUsize,
last_get_no_lock: AtomicBool,
last_put_no_lock: AtomicBool,
last_put_preconditions: Mutex<Option<HTTPPreconditions>>,
revision: AtomicUsize,
@@ -5080,6 +5092,7 @@ mod tests {
heal_replacement,
heal_calls: AtomicUsize::new(0),
write_calls: AtomicUsize::new(0),
last_get_no_lock: AtomicBool::new(false),
last_put_no_lock: AtomicBool::new(false),
last_put_preconditions: Mutex::new(None),
revision: AtomicUsize::new(1),
@@ -5135,8 +5148,9 @@ mod tests {
_object: &str,
_range: Option<HTTPRangeSpec>,
_h: HeaderMap,
_opts: &ObjectOptions,
opts: &ObjectOptions,
) -> Result<GetObjectReader> {
self.last_get_no_lock.store(opts.no_lock, Ordering::SeqCst);
let data = match &*self.state.lock().expect("state lock poisoned") {
RecoveryReadState::Missing => return Err(Error::ConfigNotFound),
RecoveryReadState::Blob(data) => data.clone(),
@@ -5431,6 +5445,60 @@ mod tests {
assert!(data.is_empty());
}
#[tokio::test]
async fn read_config_preserve_empty_metadata_keeps_empty_and_options() {
let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));
for no_lock in [false, true] {
let opts = ObjectOptions {
no_lock,
..Default::default()
};
let (data, info) = read_config_preserve_empty_with_metadata_opts(store.clone(), "config/empty.json", &opts)
.await
.expect("an existing empty config must retain its payload and metadata");
assert!(data.is_empty());
assert_eq!(info.size, 0);
assert_eq!(info.actual_size, 0);
assert_eq!(info.etag.as_deref(), Some("config-1"));
assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
assert_eq!(store.last_get_no_lock.load(Ordering::SeqCst), no_lock);
}
let err = read_config(store, "config/empty.json")
.await
.expect_err("the existing reader must still treat empty configs as missing");
assert!(matches!(err, Error::ConfigNotFound));
}
#[tokio::test]
async fn read_config_preserve_empty_metadata_keeps_payload_and_metadata() {
let payload = b"original config bytes".to_vec();
let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(payload.clone()), None));
let (data, info) =
read_config_preserve_empty_with_metadata_opts(store, "config/nonempty.json", &ObjectOptions::default())
.await
.expect("a nonempty config must retain its payload and metadata");
assert_eq!(data, payload);
let size = i64::try_from(payload.len()).expect("test payload size fits i64");
assert_eq!(info.size, size);
assert_eq!(info.actual_size, size);
assert_eq!(info.etag.as_deref(), Some("config-1"));
assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
}
#[tokio::test]
async fn read_config_preserve_empty_metadata_preserves_read_errors() {
let missing = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None));
let err = read_config_preserve_empty_with_metadata_opts(missing, "config/missing.json", &ObjectOptions::default())
.await
.expect_err("physical absence must remain ConfigNotFound");
assert!(matches!(err, Error::ConfigNotFound));
let quorum = Arc::new(RecoveryMockStore::new(RecoveryReadState::QuorumError, None));
let err = read_config_preserve_empty_with_metadata_opts(quorum, "config/quorum.json", &ObjectOptions::default())
.await
.expect_err("a quorum failure must retain its typed read error");
assert!(matches!(err, Error::ErasureReadQuorum));
}
#[async_trait::async_trait]
impl StorageAdminApi for RecoveryMockStore {
type BackendInfo = rustfs_madmin::BackendInfo;
+1 -1
View File
@@ -2911,7 +2911,7 @@ mod tests {
let timed = wrapper.track_disk_health_with_op(
"write_all",
|| std::future::pending::<Result<()>>(),
std::future::pending::<Result<()>>,
SLOW_DISK_OPERATION_THRESHOLD + Duration::from_secs(1),
);
tokio::pin!(timed);
+221
View File
@@ -12477,6 +12477,227 @@ mod test {
}
}
// SIGKILL tests process death on a healthy filesystem, not power loss.
#[cfg(unix)]
#[tokio::test]
async fn rename_data_process_kill_preserves_publication_boundary() {
use std::io::Write as _;
use std::os::unix::process::ExitStatusExt;
use std::process::{Child, Command, ExitStatus, Stdio};
use std::time::{Duration, Instant};
const TEST: &str = "disk::local::test::rename_data_process_kill_preserves_publication_boundary";
const ROLE: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROLE";
const ROOT: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROOT";
const NONCE: &str = "RUSTFS_TEST_PUBLICATION_KILL_NONCE";
const BUCKET: &str = "bucket";
const OBJECT: &str = "process-kill-object";
const STAGED: &str = "process-kill-staged";
const VERSION: Uuid = Uuid::from_u128(0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa);
const OLD: Uuid = Uuid::from_u128(0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb);
const NEW: Uuid = Uuid::from_u128(0xcccccccccccccccccccccccccccccccc);
struct ChildGuard(Child);
impl Drop for ChildGuard {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
async fn wait_exit(child: &mut ChildGuard) -> ExitStatus {
let deadline = Instant::now() + Duration::from_secs(60);
loop {
if child.0.try_wait().expect("poll owned child").is_some() {
return child.0.wait().expect("reap owned child");
}
assert!(Instant::now() < deadline, "owned child exceeded exit deadline");
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
fn spawn(root: &std::path::Path, nonce: &str, role: &str) -> ChildGuard {
let stdout = std::fs::File::create(root.join(format!("{role}.stdout"))).expect("child stdout file");
let stderr = std::fs::File::create(root.join(format!("{role}.stderr"))).expect("child stderr file");
ChildGuard(
Command::new(std::env::current_exe().expect("test executable"))
.args(["--exact", TEST, "--nocapture"])
.env(ROLE, role)
.env(ROOT, root)
.env(NONCE, nonce)
.stdout(Stdio::from(stdout))
.stderr(Stdio::from(stderr))
.spawn()
.expect("spawn owned publication child"),
)
}
fn child_logs(root: &std::path::Path, role: &str) -> String {
let stdout = std::fs::read_to_string(root.join(format!("{role}.stdout"))).expect("child stdout");
let stderr = std::fs::read_to_string(root.join(format!("{role}.stderr"))).expect("child stderr");
format!("{stdout} {stderr}")
}
fn file_info(data_dir: Uuid) -> FileInfo {
let mut fi = test_file_info(OBJECT, VERSION, Some(data_dir), None);
fi.size = 8;
fi.parts[0].size = 8;
fi.parts[0].actual_size = 8;
fi
}
if let Some(role) = std::env::var_os(ROLE) {
let role = role.to_str().expect("child role is UTF-8");
assert!(matches!(role, "writer-before" | "writer-after" | "verify-before" | "verify-after"));
let root = std::path::PathBuf::from(std::env::var_os(ROOT).expect("private child root"));
let nonce = std::env::var(NONCE).expect("private child nonce");
Uuid::parse_str(&nonce).expect("nonce is a UUID");
assert_eq!(root.file_name().and_then(|name| name.to_str()), Some(nonce.as_str()));
assert_eq!(std::fs::read_to_string(root.join("control-nonce")).expect("bound root nonce"), nonce);
let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
let _strict = durability_mode_override::set(DurabilityMode::Strict);
let disk = LocalDisk::new(&endpoint, false).await.expect("child opens real disk");
let object_dir = root.join(BUCKET).join(OBJECT);
if role.starts_with("verify-") {
let fi = disk
.read_version("", BUCKET, OBJECT, &VERSION.to_string(), &ReadOptions::default())
.await
.expect("fresh verifier reads actual version");
let (expected_dir, expected_bytes) = if role == "verify-before" {
assert_eq!(
fs::read(object_dir.join(STORAGE_FORMAT_FILE))
.await
.expect("read old metadata"),
fs::read(root.join("old-meta")).await.expect("read old fixture"),
"prepublication metadata must remain byte-for-byte old"
);
(OLD, b"old-data")
} else {
(NEW, b"new-data")
};
assert_eq!(fi.data_dir, Some(expected_dir), "{role}: actual reopened version data_dir");
let part = object_dir
.join(fi.data_dir.expect("referenced data dir").to_string())
.join("part.1");
assert_eq!(fs::read(part).await.expect("read referenced shard"), expected_bytes);
return;
}
let staged = root.join(RUSTFS_META_TMP_BUCKET).join(STAGED).join(NEW.to_string());
fs::create_dir_all(&staged).await.expect("staged shard directory");
fs::write(staged.join("part.1"), b"new-data")
.await
.expect("stage new shard bytes");
let ready = root.join(format!("{role}.ready"));
let pause = move || {
let pending = ready.with_extension("pending");
std::fs::write(&pending, &nonce).expect("write actual writer cut marker");
std::fs::rename(pending, &ready).expect("announce complete writer cut marker");
let (_sender, receiver) = std::sync::mpsc::channel::<()>();
receiver.recv().expect("writer remains paused until SIGKILL");
};
let _hook = if role == "writer-before" {
// Use the executor's exact path spelling, including canonical
// macOS roots and Linux mount-pinned I/O roots.
let destination = disk
.io_get_object_path(BUCKET, &format!("{OBJECT}/{STORAGE_FORMAT_FILE}"))
.expect("actual metadata publication path");
assert_eq!(
std::fs::canonicalize(&destination).expect("canonical publication destination"),
disk.root.join(BUCKET).join(OBJECT).join(STORAGE_FORMAT_FILE),
"hook destination must identify this disk's seeded metadata"
);
Some(os::prepared_publication_test_hooks::install_at(
os::prepared_publication_test_hooks::Stage::PreparedRename,
&destination,
pause,
))
} else {
disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
.await
.expect("Strict real rename_data completes before cut");
pause();
return;
};
disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
.await
.expect("before writer must remain in physical hook");
panic!("prepublication hook unexpectedly returned");
}
let temp = tempfile::tempdir().expect("parent owns private disk roots");
for phase in ["before", "after"] {
let nonce = Uuid::new_v4().to_string();
let root = temp.path().join(&nonce);
fs::create_dir(&root).await.expect("private scenario root");
fs::write(root.join("control-nonce"), &nonce)
.await
.expect("bind private root");
let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("seed real disk");
ensure_test_volume(&disk, BUCKET).await;
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
let object_dir = root.join(BUCKET).join(OBJECT);
fs::create_dir_all(object_dir.join(OLD.to_string()))
.await
.expect("old shard directory");
fs::write(object_dir.join(OLD.to_string()).join("part.1"), b"old-data")
.await
.expect("seed old shard bytes");
let old_meta = test_meta(file_info(OLD));
fs::write(object_dir.join(STORAGE_FORMAT_FILE), &old_meta)
.await
.expect("seed old metadata");
fs::write(root.join("old-meta"), &old_meta)
.await
.expect("retain exact old fixture");
drop(disk);
let writer_role = format!("writer-{phase}");
let mut writer = spawn(&root, &nonce, &writer_role);
let deadline = Instant::now() + Duration::from_secs(60);
let ready = root.join(format!("{writer_role}.ready"));
loop {
if let Some(status) = writer.0.try_wait().expect("poll writer") {
panic!(
"writer exited before cut: {writer_role} status={status}: {}",
child_logs(&root, &writer_role)
);
}
match fs::read_to_string(&ready).await {
Ok(marker) => {
assert_eq!(marker, nonce, "cut marker belongs to this private writer");
break;
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => panic!("read cut marker: {error}"),
}
if Instant::now() >= deadline {
panic!("writer cut deadline: {writer_role}: {}", child_logs(&root, &writer_role));
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let pid = writer.0.id();
writer.0.kill().expect("SIGKILL owned writer at verified cut");
let status = wait_exit(&mut writer).await;
assert_eq!(status.signal(), Some(9), "owned writer must actually die by SIGKILL");
writeln!(std::io::stdout(), "{writer_role} pid={pid} signal={:?}", status.signal())
.expect("record actual writer wait status");
let verifier_role = format!("verify-{phase}");
let mut verifier = spawn(&root, &nonce, &verifier_role);
let status = wait_exit(&mut verifier).await;
let logs = child_logs(&root, &verifier_role);
assert!(status.success(), "{verifier_role} status={status}: {logs}");
let stdout = std::fs::read_to_string(root.join(format!("{verifier_role}.stdout"))).expect("verifier stdout");
assert!(
stdout.contains(TEST) && stdout.contains("1 passed"),
"verifier must execute exact test: {logs}"
);
writeln!(std::io::stdout(), "{verifier_role} status={status}").expect("record actual verifier exit");
}
}
/// Crash-consistency harness for the rename_data commit sequence
/// (rustfs/backlog#935 HP-14, test plan rustfs/backlog#896; hard rule from
/// rustfs/backlog#878: "After partial commit, the object can only be an old or new version; it cannot be mixed").
+53 -10
View File
@@ -442,12 +442,19 @@ impl WritePlan {
}
if let Some(encryption) = self.encryption {
let etag_nonce = match encryption.mode {
WriteEncryptionMode::Singlepart { base_nonce } | WriteEncryptionMode::MultipartLegacy { base_nonce, .. } => {
base_nonce
}
WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
};
let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
reader = match encryption.mode {
WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
#[cfg(feature = "rio-v2")]
EncryptReader::new_with_object_key(reader, encryption.key_bytes),
EncryptReader::new_with_object_key(protected_reader, encryption.key_bytes),
#[cfg(not(feature = "rio-v2"))]
EncryptReader::new(reader, encryption.key_bytes, [0u8; 12]),
EncryptReader::new(protected_reader, encryption.key_bytes, [0u8; 12]),
HashReader::SIZE_PRESERVE_LAYER,
actual_size,
None,
@@ -457,12 +464,12 @@ impl WritePlan {
WriteEncryptionMode::Singlepart { base_nonce } => {
#[cfg(not(feature = "rio-v2"))]
let encrypt_reader = if encryption_frame_v2_enabled() {
EncryptReader::new_v2(reader, encryption.key_bytes, base_nonce)
EncryptReader::new_v2(protected_reader, encryption.key_bytes, base_nonce)
} else {
EncryptReader::new(reader, encryption.key_bytes, base_nonce)
EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce)
};
#[cfg(feature = "rio-v2")]
let encrypt_reader = EncryptReader::new(reader, encryption.key_bytes, base_nonce);
let encrypt_reader = EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce);
HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
}
WriteEncryptionMode::MultipartLegacy {
@@ -471,20 +478,25 @@ impl WritePlan {
} => {
#[cfg(not(feature = "rio-v2"))]
let encrypt_reader = if encryption_frame_v2_enabled() {
EncryptReader::new_multipart_v2(reader, encryption.key_bytes, base_nonce, multipart_part_number)
EncryptReader::new_multipart_v2(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
} else {
EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number)
EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
};
#[cfg(feature = "rio-v2")]
let encrypt_reader =
EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number);
EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number);
HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
}
WriteEncryptionMode::MultipartObjectKey { multipart_part_number } => HashReader::from_reader(
#[cfg(feature = "rio-v2")]
EncryptReader::new_multipart_with_object_key(reader, encryption.key_bytes, multipart_part_number),
EncryptReader::new_multipart_with_object_key(protected_reader, encryption.key_bytes, multipart_part_number),
#[cfg(not(feature = "rio-v2"))]
EncryptReader::new_multipart(reader, encryption.key_bytes, [0u8; 12], multipart_part_number as usize),
EncryptReader::new_multipart(
protected_reader,
encryption.key_bytes,
[0u8; 12],
multipart_part_number as usize,
),
HashReader::SIZE_PRESERVE_LAYER,
actual_size,
None,
@@ -497,6 +509,7 @@ impl WritePlan {
// `ignore_value` deliberately avoids a second hasher over compressed or
// encrypted bytes. The inner reader still validates the plaintext request
// checksum while this outer reader exposes the request checksum context.
reader.encrypted_etag = encrypted;
reader.add_non_trailing_checksum(checksum, true)?;
reader.set_trailer(trailer);
@@ -513,6 +526,36 @@ mod tests {
use std::io::Cursor;
use tokio::io::AsyncReadExt;
#[tokio::test]
async fn write_plan_encryption_persists_protected_etags_for_every_mode() {
let data = b"Salary 2026: 85000 EUR";
let size = i64::try_from(data.len()).expect("test data size");
let mut control =
HashReader::from_stream(Cursor::new(data), size, size, None, None, false).expect("create control reader");
control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
let md5 = control.try_resolve_etag().expect("control MD5");
for encryption in [
WriteEncryption::singlepart([1; 32], [1; 12]),
WriteEncryption::singlepart_object_key([1; 32]),
WriteEncryption::multipart([1; 32], [1; 12], 1),
WriteEncryption::multipart_object_key([1; 32], 1),
] {
let plaintext = HashReader::from_stream(Cursor::new(data), size, size, Some(md5.clone()), None, false)
.expect("create checked reader");
let mut reader = WritePlan::new()
.with_encryption(encryption)
.apply(plaintext, size)
.expect("apply encryption");
let mut ciphertext = Vec::new();
reader.read_to_end(&mut ciphertext).await.expect("stream ciphertext");
assert_ne!(ciphertext, data);
let tag = reader.try_resolve_etag().expect("resolve persisted encrypted ETag");
assert_ne!(tag, md5, "stored ETag must not reveal the plaintext MD5");
assert_eq!(tag.len(), 32, "multipart ETags require a 16-byte hexadecimal part tag");
assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
}
}
async fn assert_non_trailing_checksum_survives(plan: WritePlan) {
let plaintext = b"checksum-context-through-write-plan".repeat(256);
let actual_size = plaintext.len() as i64;
@@ -3004,6 +3004,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// etag
let etag = resolve_complete_etag(opts, &uploaded_parts);
// An upload can contain parts from an older writer or a peer. Until
// every part's format is attested, keep its ETag out of keyless lists.
rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
fi.metadata.insert("etag".to_owned(), etag);
let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata);
+69
View File
@@ -3847,6 +3847,16 @@ impl SetDisks {
.map(crate::io_support::rio::compression_index_storage_bytes);
let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
if data.stream.encrypted_etag {
if opts.preserve_etag.as_ref().is_none_or(|preserved| preserved == &etag) {
insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
} else {
rustfs_utils::http::metadata_compat::remove_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
);
}
}
if let Some(ref tag) = opts.preserve_etag {
etag = tag.clone();
}
@@ -10610,6 +10620,65 @@ mod object_encryption_resolver_wiring_tests {
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
#[tokio::test]
async fn encrypted_put_attests_only_etags_produced_by_its_encryption_reader() {
use super::hermetic_set_disks_support::hermetic_set_disks_isolated;
use crate::io_support::rio::{WriteEncryption, WritePlan};
use crate::storage_api_contracts::object::ObjectIO as _;
use tokio::io::AsyncReadExt;
let (_temp_dirs, disks, set_disks) = hermetic_set_disks_isolated(4).await;
let bucket = "encrypted-etag-proof";
for disk in &disks {
disk.make_volume(bucket).await.expect("create bucket volume");
}
let plaintext = b"Salary 2026: 85000 EUR";
let size = plaintext.len() as i64;
let mut control = HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("control reader");
control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
let plaintext_md5 = control.try_resolve_etag().expect("control MD5");
for (object, preserved) in [("new", None), ("preserved", Some(plaintext_md5.clone()))] {
let reader =
HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("plaintext reader");
let reader = WritePlan::new()
.with_encryption(WriteEncryption::singlepart([1; 32], [2; 12]))
.apply(reader, size)
.expect("encrypted reader");
let mut opts = ObjectOptions {
preserve_etag: preserved.clone(),
user_defined: HashMap::from([
("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned()),
(rustfs_utils::http::SSEC_ORIGINAL_SIZE_HEADER.to_owned(), size.to_string()),
]),
..Default::default()
};
// A copied producer marker cannot attest to a preserved plaintext tag.
insert_str(&mut opts.user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
set_disks
.put_object(bucket, object, &mut PutObjReader::new(reader), &opts)
.await
.expect("store encrypted bytes");
let stored = set_disks
.get_object_info(bucket, object, &ObjectOptions::default())
.await
.expect("read persisted metadata");
if let Some(preserved) = preserved {
assert_eq!(stored.etag.as_deref(), Some(preserved.as_str()));
assert!(!rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
} else {
assert_ne!(stored.etag.as_ref(), Some(&plaintext_md5));
assert_eq!(
rustfs_utils::http::get_consistent_str(
&stored.user_defined,
rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG
),
Some("v1")
);
assert!(rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
}
}
}
#[derive(Clone, Default)]
struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
+193 -15
View File
@@ -901,6 +901,7 @@ impl ECStore {
}
};
let confirmed_missing = existing_bucket_info.is_none();
let mut pending_creation_metadata = None;
if confirmed_missing
&& !opts.no_lock
&& !is_meta_bucketname(bucket)
@@ -921,13 +922,25 @@ impl ECStore {
),
other => other,
})?;
return Err(StorageError::InvalidArgument(
"CreateBucket".to_string(),
"bucket".to_string(),
format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
));
let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
let metadata_store = metadata_sys::object_store_in(&self.ctx).await?;
let pending_intent = persisted
&& metadata.bucket_incarnation_id == old_incarnation
&& metadata.needs_bucket_creation_commit()
&& !crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?;
if pending_intent {
pending_creation_metadata = Some(metadata);
} else {
return Err(StorageError::InvalidArgument(
"CreateBucket".to_string(),
"bucket".to_string(),
format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
));
}
}
let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
let existing_metadata = if let Some(metadata) = pending_creation_metadata {
Some(metadata)
} else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
if !persisted {
metadata = BucketMetadata::new(bucket);
@@ -1347,6 +1360,23 @@ impl ECStore {
} else {
None
};
// Match metadata migration: lifecycle -> metadata transaction ->
// bucket namespace. Holding the transaction while waiting for the
// namespace prevents a migration writer and force delete from
// acquiring these two locks in opposite orders.
let object_lock_metadata_transaction = if opts.force && !is_meta_bucketname(bucket) {
Some(
await_bucket_namespace_operation(
bucket_lifecycle_guard.as_ref(),
bucket,
"Object Lock metadata transaction acquisition",
crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket),
)
.await?,
)
} else {
None
};
let ns_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, bucket).await?;
Some(
@@ -1402,6 +1432,29 @@ impl ECStore {
}
}
// Keep configuration changes fenced until data deletion finishes;
// a force delete cannot bypass any Object Lock protection.
let object_lock_metadata_guard = if bucket_exists && let Some(guard) = object_lock_metadata_transaction {
let (state, incarnation, _) =
crate::bucket::metadata_sys::get_object_lock_config_and_incarnation_from_disk_with_guard_in(
&self.ctx,
bucket,
Some(ns_guard.as_ref().unwrap_or(&guard)),
)
.await?;
match state {
crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent => Some((guard, incarnation)),
crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. } => {
return Err(StorageError::PrefixAccessDenied(bucket.to_owned(), String::new()));
}
crate::bucket::metadata_sys::ObjectLockConfigState::Fabricated => {
return Err(StorageError::FileCorrupt);
}
}
} else {
None
};
if sr_delete && !bucket_exists {
delete_opts.force_if_empty = true;
}
@@ -1415,7 +1468,12 @@ impl ECStore {
ns_guard.as_ref(),
bucket,
"bucket delete marker creation",
self.mark_bucket_deleted(bucket),
await_bucket_namespace_operation(
object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
bucket,
"Object Lock configuration fence during bucket retirement",
self.mark_bucket_deleted(bucket),
),
)
.await?;
}
@@ -1424,9 +1482,13 @@ impl ECStore {
// Legacy buckets without a stamp cannot produce retirement authority.
let retirement = if bucket_exists && bucket_lifecycle_guard.is_some() && !is_meta_bucketname(bucket) {
if let Some(store) = metadata_sys::object_store_if_initialized_in(&self.ctx).await {
crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
.await?
.map(|incarnation| (store, incarnation))
if let Some((_, incarnation)) = object_lock_metadata_guard.as_ref() {
Some((store, *incarnation))
} else {
crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
.await?
.map(|incarnation| (store, incarnation))
}
} else {
None
}
@@ -1438,13 +1500,19 @@ impl ECStore {
bucket_lifecycle_guard.as_ref(),
bucket,
"physical bucket deletion",
run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
self.delete_bucket_on_sets(bucket, &delete_opts)
.await
.map_err(|err| to_object_err(err, vec![bucket]))
}),
await_bucket_namespace_operation(
object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
bucket,
"Object Lock configuration fence during bucket deletion",
run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
self.delete_bucket_on_sets(bucket, &delete_opts)
.await
.map_err(|err| to_object_err(err, vec![bucket]))
}),
),
)
.await;
drop(object_lock_metadata_guard);
if let Err(err) = delete_result
&& (!sr_delete || !is_err_strict_volume_not_found(&err))
{
@@ -3689,6 +3757,116 @@ mod tests {
assert!(!any_disk_path_exists(&disk_paths, bucket_deleted_marker_volume(&bucket)).await);
}
#[tokio::test]
#[serial]
async fn force_bucket_delete_preserves_object_lock_versions() {
let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], Some(2)).await;
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("force-delete-worm-{}", Uuid::new_v4().simple());
store
.make_bucket(
&bucket,
&MakeBucketOptions {
lock_enabled: true,
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("create Object Lock bucket");
let until = (OffsetDateTime::now_utc() + time::Duration::days(365))
.format(&time::format_description::well_known::Rfc3339)
.expect("format future retention date");
let written = store
.put_object(
&bucket,
"retained",
&mut PutObjReader::from_vec(b"retained bytes".to_vec()),
&ObjectOptions {
versioned: true,
user_defined: HashMap::from([
("x-amz-object-lock-mode".to_owned(), "COMPLIANCE".to_owned()),
("x-amz-object-lock-retain-until-date".to_owned(), until),
]),
..Default::default()
},
)
.await
.expect("write retained object");
for operation in [SRBucketDeleteOp::NoOp, SRBucketDeleteOp::MarkDelete, SRBucketDeleteOp::Purge] {
let err = store
.delete_bucket(
&bucket,
&DeleteBucketOptions {
force: true,
srdelete_op: operation,
..Default::default()
},
)
.await
.expect_err("force deletion must not bypass Object Lock");
assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected denial: {err}");
let info = store
.get_object_info(
&bucket,
"retained",
&ObjectOptions {
version_id: written.version_id.map(|version| version.to_string()),
..Default::default()
},
)
.await
.expect("retained version must survive the rejected bucket deletion");
assert_eq!(info.etag, written.etag);
assert_eq!(info.size, written.size);
assert_eq!(info.version_id, written.version_id);
}
metadata_sys::inject_object_lock_disk_read_error_in(&store.ctx, &bucket)
.await
.expect("inject authoritative metadata read failure");
let error = store
.delete_bucket(
&bucket,
&DeleteBucketOptions {
force: true,
..Default::default()
},
)
.await
.expect_err("an unreadable Object Lock configuration must fail closed");
assert!(error.to_string().contains("injected Object Lock metadata disk read failure"));
let info = store
.get_object_info(&bucket, "retained", &ObjectOptions::default())
.await
.expect("metadata read failure must leave the retained object intact");
assert_eq!(info.version_id, written.version_id);
let unlocked = format!("force-delete-unlocked-{}", Uuid::new_v4().simple());
store
.make_bucket(&unlocked, &MakeBucketOptions::default())
.await
.expect("create unlocked control bucket");
store
.put_object(
&unlocked,
"ordinary",
&mut PutObjReader::from_vec(b"ordinary bytes".to_vec()),
&ObjectOptions::default(),
)
.await
.expect("write ordinary control object");
store
.delete_bucket(
&unlocked,
&DeleteBucketOptions {
force: true,
..Default::default()
},
)
.await
.expect("force deletion must still work for an unlocked bucket");
}
#[tokio::test]
#[serial]
async fn bucket_delete_mark_delete_rejects_hidden_object_paths_without_force() {
+1 -1
View File
@@ -18318,7 +18318,7 @@ mod tests {
let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4());
let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
intent.lock_enabled = true;
crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
.await
.expect("persist pre-visibility lock intent");
assert!(
+116 -52
View File
@@ -448,8 +448,8 @@ const MAX_LIST_OBJECTS_METADATA_FAST_STALENESS_MS: u64 = 60_000;
const LIST_OBJECTS_INDEX_PROVIDER_WALKER_KEY_ONLY: &str = "walker_key_only";
const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY: &str = "persistent_key_only";
const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY_DEFAULT_GENERATION: &str = "persistent-key-only";
const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 2;
const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v2";
const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 3;
const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v3";
const PERSISTENT_KEY_ONLY_INDEX_BUCKET_HEADER: &str = "# bucket=";
const PERSISTENT_KEY_ONLY_INDEX_GENERATION_HEADER: &str = "# generation=";
const PERSISTENT_KEY_ONLY_INDEX_CHECKPOINT_HEADER: &str = "# checkpoint_high_water_mark=";
@@ -656,7 +656,10 @@ impl PersistentListMetadataObject {
name: object.name.clone(),
size: object.size,
mod_time: object.mod_time,
etag: object.etag.clone(),
etag: object
.etag
.clone()
.filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined)),
storage_class: object.storage_class.clone(),
}
}
@@ -2361,24 +2364,43 @@ fn list_objects_paginate(
(objects, prefixes, is_truncated, next_marker, next_version_idmarker)
}
fn list_objects_paginate_versions<'a>(
mut get_objects: Vec<ObjectInfo>,
delimiter: &Option<String>,
struct VersionedListPagination<'a, I> {
delimiter: &'a Option<String>,
max_keys: i32,
disk_has_more: bool,
cache_id: Option<&str>,
marker: Option<&str>,
last_scanned_key: Option<&str>,
raw_keys: impl Iterator<Item = &'a str>,
) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>) {
filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker);
let prefix_at_page_boundary =
versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more);
cache_id: Option<&'a str>,
marker: Option<&'a str>,
last_scanned_key: Option<&'a str>,
raw_keys: I,
}
let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) =
list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key);
fn list_objects_paginate_versions<'a, I>(
mut get_objects: Vec<ObjectInfo>,
pagination: VersionedListPagination<'a, I>,
) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>)
where
I: Iterator<Item = &'a str>,
{
filter_versions_common_prefixes_after_marker(&mut get_objects, pagination.delimiter.as_deref(), pagination.marker);
let prefix_at_page_boundary = versioned_prefix_page_boundary_key(
&get_objects,
pagination.raw_keys,
pagination.delimiter.as_deref(),
pagination.max_keys,
pagination.disk_has_more,
);
let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = list_objects_paginate(
get_objects,
pagination.delimiter,
pagination.max_keys,
pagination.disk_has_more,
pagination.cache_id,
true,
pagination.last_scanned_key,
);
if is_truncated && let Some(raw_key) = prefix_at_page_boundary {
next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id));
next_marker = Some(append_list_cache_id_to_marker(raw_key, pagination.cache_id));
next_version_idmarker = None;
}
@@ -4326,13 +4348,15 @@ impl ECStore {
let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
get_objects,
&delimiter,
max_keys,
disk_has_more,
next_cache_id.as_deref(),
opts.marker.as_deref(),
last_scanned_key.as_deref(),
entries.entries().iter().map(|entry| entry.name.as_str()),
VersionedListPagination {
delimiter: &delimiter,
max_keys,
disk_has_more,
cache_id: next_cache_id.as_deref(),
marker: opts.marker.as_deref(),
last_scanned_key: last_scanned_key.as_deref(),
raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
},
);
Ok(ListObjectVersionsInfo {
@@ -5754,13 +5778,15 @@ impl Sets {
let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
get_objects,
&delimiter,
max_keys,
disk_has_more,
next_cache_id.as_deref(),
opts.marker.as_deref(),
last_scanned_key.as_deref(),
entries.entries().iter().map(|entry| entry.name.as_str()),
VersionedListPagination {
delimiter: &delimiter,
max_keys,
disk_has_more,
cache_id: next_cache_id.as_deref(),
marker: opts.marker.as_deref(),
last_scanned_key: last_scanned_key.as_deref(),
raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
},
);
Ok(ListObjectVersionsInfo {
@@ -6568,13 +6594,15 @@ impl SetDisks {
let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
get_objects,
&delimiter,
max_keys,
disk_has_more,
next_cache_id.as_deref(),
opts.marker.as_deref(),
last_scanned_key.as_deref(),
entries.entries().iter().map(|entry| entry.name.as_str()),
VersionedListPagination {
delimiter: &delimiter,
max_keys,
disk_has_more,
cache_id: next_cache_id.as_deref(),
marker: opts.marker.as_deref(),
last_scanned_key: last_scanned_key.as_deref(),
raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
},
);
Ok(ListObjectVersionsInfo {
@@ -8848,13 +8876,15 @@ mod test {
let get_objects = fold_delimiter_page(&window, "", "-");
let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions(
get_objects,
&delimiter,
max_keys,
disk_has_more,
None,
marker.as_deref(),
last_scanned.as_deref(),
window.iter().map(String::as_str),
super::VersionedListPagination {
delimiter: &delimiter,
max_keys,
disk_has_more,
cache_id: None,
marker: marker.as_deref(),
last_scanned_key: last_scanned.as_deref(),
raw_keys: window.iter().map(String::as_str),
},
);
assert!(
@@ -8927,13 +8957,15 @@ mod test {
let get_objects = fold_delimiter_page(&window, "", "-");
let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions(
get_objects,
&delimiter,
max_keys,
disk_has_more,
None,
marker.as_deref(),
last_scanned.as_deref(),
window.iter().map(String::as_str),
super::VersionedListPagination {
delimiter: &delimiter,
max_keys,
disk_has_more,
cache_id: None,
marker: marker.as_deref(),
last_scanned_key: last_scanned.as_deref(),
raw_keys: window.iter().map(String::as_str),
},
);
assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys");
@@ -9176,6 +9208,38 @@ mod test {
assert_eq!(parsed, object);
}
#[test]
fn persistent_listing_snapshots_cannot_restore_legacy_encrypted_fingerprints() {
let mut object = ObjectInfo {
name: "encrypted".to_owned(),
etag: Some("fbfcbcc6d2f035411e6268c8b119593e".to_owned()),
user_defined: Arc::new(HashMap::from([(
"x-amz-server-side-encryption-customer-algorithm".to_owned(),
"AES256".to_owned(),
)])),
..Default::default()
};
let snapshot = PersistentListMetadataObject::from_object_info(&object);
assert!(snapshot.to_object_info("bucket").etag.is_none());
let mut old_snapshot = snapshot.clone();
old_snapshot.etag = object.etag.clone();
let old_contents = format!(
"# rustfs-listobjects-key-only-v2\n{}\n",
encode_persistent_list_metadata_object(&old_snapshot)
);
let old_index = parse_persistent_key_only_index(&old_contents);
assert_ne!(old_index.format_version, PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION);
rustfs_utils::http::metadata_compat::insert_str(
Arc::make_mut(&mut object.user_defined),
rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
"v1".to_owned(),
);
assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
object.user_defined = Arc::new(HashMap::new());
assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
}
#[test]
fn persistent_key_only_index_parses_legacy_keys_and_metadata_snapshots() {
let object = PersistentListMetadataObject {
+83 -3
View File
@@ -5351,13 +5351,41 @@ impl ECStore {
tier_journal_api: Option<Arc<ECStore>>,
) -> (Vec<DeletedObject>, Vec<Option<Error>>, Vec<Option<DeleteAccounting>>) {
Box::pin(async move {
let count = objects.len();
let mut deleted = vec![DeletedObject::default(); count];
let mut errors: Vec<Option<Error>> = (0..count).map(|_| None).collect();
let mut accounting = vec![None; count];
let mut valid_objects = Vec::with_capacity(count);
let mut valid_indices = Vec::with_capacity(count);
// Validate the literal key before encoding it or taking locks. A
// dot-segment key must never reach filesystem path resolution.
for (index, object) in objects.into_iter().enumerate() {
match check_del_obj_args(bucket, &object.object_name) {
Ok(()) => {
valid_indices.push(index);
valid_objects.push(object);
}
Err(err) => errors[index] = Some(err),
}
}
if valid_objects.is_empty() {
return (deleted, errors, accounting);
}
let mut opts = opts;
let receipt_sink = install_tier_free_version_receipt_sink(&mut opts);
let result = self
.handle_delete_objects_with_journal_and_accounting_inner(bucket, objects, opts, tier_journal_api)
let (valid_deleted, valid_errors, valid_accounting) = self
.handle_delete_objects_with_journal_and_accounting_inner(bucket, valid_objects, opts, tier_journal_api)
.await;
enqueue_recorded_tier_free_versions(self, receipt_sink).await;
result
for (index, ((object, error), receipt)) in valid_indices
.into_iter()
.zip(valid_deleted.into_iter().zip(valid_errors).zip(valid_accounting))
{
deleted[index] = object;
errors[index] = error;
accounting[index] = receipt;
}
(deleted, errors, accounting)
})
.await
}
@@ -9094,6 +9122,58 @@ mod tests {
(dirs, store)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn batch_delete_rejects_dot_segments_and_keeps_valid_result_positions() {
let bucket = "batch-delete-key-validation";
let (_dirs, store) = multipool_version_test_store(bucket).await;
let options = ObjectOptions {
versioned: true,
..Default::default()
};
let victim = store
.put_object(bucket, "victim/object", &mut PutObjReader::from_vec(b"victim".to_vec()), &options)
.await
.expect("write protected-prefix control object");
store
.put_object(bucket, "allowed/object", &mut PutObjReader::from_vec(b"allowed".to_vec()), &options)
.await
.expect("write valid deletion control object");
let keys = [
"allowed/../victim/object",
"allowed/object",
"a/./b",
"../other-bucket/object",
];
let objects = keys
.iter()
.map(|key| ObjectToDelete {
object_name: (*key).to_owned(),
..Default::default()
})
.collect();
let (deleted, errors) = store.handle_delete_objects(bucket, objects, options).await;
assert_eq!(deleted.len(), keys.len());
assert_eq!(errors.len(), keys.len());
for index in [0, 2, 3] {
assert!(
matches!(&errors[index], Some(StorageError::ObjectNameInvalid(_, _))),
"invalid key {}: {:?}",
keys[index],
errors[index]
);
assert!(!deleted[index].delete_marker);
}
assert!(errors[1].is_none(), "valid deletion must still succeed: {:?}", errors[1]);
assert!(deleted[1].delete_marker);
assert_eq!(deleted[1].object_name, "allowed/object");
let after = store
.get_object_info(bucket, "victim/object", &ObjectOptions::default())
.await
.expect("the resolved victim path must remain visible");
assert_eq!(after.version_id, victim.version_id);
assert_eq!(after.etag, victim.etag);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn retired_marker_converges_after_bucket_recreation_on_sixteen_disks() {
let bucket = "retired-marker-c11";
@@ -392,7 +392,7 @@ impl PartialWrites {
object: anchor.object.clone(),
version_id: anchor.version_id,
scope: anchor.scope,
delete_marker_purge: anchor.delete_marker_purge.clone(),
delete_marker_purge: anchor.delete_marker_purge,
lease: Some(anchor.lease),
bucket_incarnation_id: anchor.bucket_incarnation_id,
disposition: MrfVerifiedRepairDisposition::AuthoritativelyAbsent,
+1
View File
@@ -30,6 +30,7 @@ workspace = true
[features]
default = []
test-util = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
+4 -6
View File
@@ -18,7 +18,6 @@ mod mapper;
mod model;
pub mod oidc;
mod provider;
mod registry;
mod transaction;
pub use binding::FederatedSessionBinding;
@@ -27,10 +26,9 @@ pub use mapper::{CoreFederatedAuthorizationMapper, FederatedAuthorizationRules};
pub(crate) use mapper::{FederatedAuthorizationRule, FederatedAuthorizationRuleRef};
pub(crate) use model::OpaqueLogoutContinuation;
pub use model::{
FederatedAuthorization, FederatedClaims, FederatedCodeExchange, FederatedLoginSession, FederatedProviderRef,
FederatedSession, FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
FederatedAuthorization, FederatedClaims, FederatedLoginSession, FederatedProviderRef, FederatedSession,
FederatedSessionTransaction, OIDC_VIRTUAL_PARENT_CLAIM, VerifiedFederatedIdentity,
};
pub use provider::{FederatedIdentityProvider, FederatedProviderView, FederatedRedirectPolicy};
pub(crate) use provider::{StandardOidcAuthentication, VerifiedFederatedCodeExchange};
pub use registry::FederatedIdentityRegistry;
pub(crate) use provider::{FederatedProviderQuery, StandardOidcAuthentication, VerifiedFederatedCodeExchange};
pub use provider::{FederatedProviderView, FederatedRedirectPolicy};
pub use transaction::FederatedIdentityService;
+2 -50
View File
@@ -164,40 +164,6 @@ impl FederatedAuthorization {
}
}
pub struct FederatedCodeExchange {
pub authorization: FederatedAuthorization,
pub redirect_after: Option<String>,
pub id_token: String,
}
impl FederatedCodeExchange {
pub(crate) fn into_parts(self) -> (FederatedAuthorization, Option<String>, OpaqueLogoutContinuation) {
let Self {
authorization,
redirect_after,
id_token,
} = self;
let continuation = OpaqueLogoutContinuation {
provider: FederatedProviderRef::new(authorization.provider_id.clone()),
id_token,
};
(authorization, redirect_after, continuation)
}
}
impl fmt::Debug for FederatedCodeExchange {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("FederatedCodeExchange")
.field("provider_id", &self.authorization.provider_id)
.field("policy_count", &self.authorization.policies.len())
.field("group_count", &self.authorization.groups.len())
.field("redirect_after_present", &self.redirect_after.is_some())
.field("id_token_present", &!self.id_token.is_empty())
.finish()
}
}
/// Carries the provider-bound ID token between code exchange and logout token creation.
pub(crate) struct OpaqueLogoutContinuation {
provider: FederatedProviderRef,
@@ -397,27 +363,13 @@ mod tests {
#[test]
fn logout_continuation_binds_provider_and_redacts_token() {
let mut authorization = authorization(Vec::new(), Vec::new());
authorization
.claims
.raw
.insert("attribute_poison".to_string(), Value::String("secret-attribute".to_string()));
let exchange = FederatedCodeExchange {
authorization,
redirect_after: Some("/console?poison=redirect-secret".to_string()),
id_token: "secret-id-token".to_string(),
};
let debug = format!("{exchange:?}");
let (_, _, continuation) = exchange.into_parts();
let continuation =
OpaqueLogoutContinuation::new(FederatedProviderRef::new("standard_oidc".to_string()), "secret-id-token".to_string());
let continuation_debug = format!("{continuation:?}");
let (provider, id_token) = continuation.into_parts();
assert_eq!(provider.as_str(), "standard_oidc");
assert_eq!(id_token, "secret-id-token");
assert!(!debug.contains("secret-id-token"));
assert!(!debug.contains("secret-attribute"));
assert!(!debug.contains("redirect-secret"));
assert!(debug.contains("id_token_present: true"));
assert!(!continuation_debug.contains("secret-id-token"));
assert!(continuation_debug.contains("[REDACTED]"));
}
+14 -41
View File
@@ -16,10 +16,9 @@ use super::config::OidcConfigQuery;
use super::{claims, config, discovery, http};
use crate::{
federation::{
CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedAuthorizationRules, FederatedCodeExchange,
FederatedIdentityProvider, FederatedIdentityService, FederatedProviderRef, FederatedProviderView,
FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
VerifiedFederatedCodeExchange,
CoreFederatedAuthorizationMapper, FederatedAuthorizationRules, FederatedIdentityService, FederatedProviderQuery,
FederatedProviderRef, FederatedProviderView, FederatedRedirectPolicy, FederationError, OpaqueLogoutContinuation, Result,
StandardOidcAuthentication, VerifiedFederatedCodeExchange,
},
oidc::{OidcConfigSnapshot, OidcSys},
};
@@ -45,8 +44,8 @@ impl StandardOidcAdapter {
pub fn into_service(self: Arc<Self>, mapper: CoreFederatedAuthorizationMapper) -> FederatedIdentityService {
let authentication: Arc<dyn StandardOidcAuthentication> = self.clone();
let provider: Arc<dyn FederatedIdentityProvider> = self;
FederatedIdentityService::from_standard_oidc_parts(provider, authentication, mapper)
let provider_query: Arc<dyn FederatedProviderQuery> = self;
FederatedIdentityService::from_standard_oidc_parts(provider_query, authentication, mapper)
}
}
@@ -58,6 +57,10 @@ impl OidcConfigQuery for StandardOidcAdapter {
#[async_trait::async_trait]
impl StandardOidcAuthentication for StandardOidcAdapter {
async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
}
async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange> {
let (oidc_claims, provider_id, session, id_token) = http::exchange_code(&self.oidc, state, code, redirect_uri).await?;
let provider = FederatedProviderRef::new(provider_id);
@@ -81,10 +84,13 @@ impl StandardOidcAuthentication for StandardOidcAdapter {
let (provider, id_token) = continuation.into_parts();
http::create_logout_token(&self.oidc, provider.as_str(), &id_token).await
}
async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
}
}
#[async_trait::async_trait]
impl FederatedIdentityProvider for StandardOidcAdapter {
impl FederatedProviderQuery for StandardOidcAdapter {
fn has_providers(&self) -> bool {
self.oidc.has_providers()
}
@@ -97,40 +103,7 @@ impl FederatedIdentityProvider for StandardOidcAdapter {
discovery::list_visible_providers(&self.oidc)
}
fn provider_config(&self, provider_id: &str) -> Option<&crate::oidc::OidcProviderConfig> {
self.oidc.get_provider_config(provider_id)
}
fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
config::redirect_policy(&self.oidc, provider_id)
}
async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
http::authorize_url(&self.oidc, provider_id, redirect_uri, redirect_after).await
}
async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange> {
let exchange = StandardOidcAuthentication::exchange_identity(self, state, code, redirect_uri).await?;
let (identity, redirect_after, continuation) = exchange.into_parts();
let authorization = CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity);
let (_, id_token) = continuation.into_parts();
Ok(FederatedCodeExchange {
authorization,
redirect_after,
id_token,
})
}
async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization> {
let identity = StandardOidcAuthentication::verify_identity(self, jwt).await?;
Ok(CoreFederatedAuthorizationMapper::new(self.authorization_rules()).map(identity))
}
async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
http::create_logout_token(&self.oidc, provider_id, id_token).await
}
async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
http::build_logout_url(&self.oidc, logout_token, post_logout_redirect_uri).await
}
}
+5 -26
View File
@@ -12,8 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::{FederatedAuthorization, FederatedCodeExchange, OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
use crate::oidc::OidcProviderConfig;
use super::{OpaqueLogoutContinuation, Result, VerifiedFederatedIdentity};
use serde::{Deserialize, Serialize};
/// Provider data exposed to login discovery consumers.
@@ -56,41 +55,21 @@ impl VerifiedFederatedCodeExchange {
#[async_trait::async_trait]
pub(crate) trait StandardOidcAuthentication: Send + Sync {
async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
async fn exchange_identity(&self, state: &str, code: &str, redirect_uri: &str) -> Result<VerifiedFederatedCodeExchange>;
async fn verify_identity(&self, jwt: &str) -> Result<VerifiedFederatedIdentity>;
async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String>;
async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
}
#[async_trait::async_trait]
pub trait FederatedIdentityProvider: Send + Sync {
pub(crate) trait FederatedProviderQuery: Send + Sync {
fn has_providers(&self) -> bool;
fn list_providers(&self) -> Vec<FederatedProviderView>;
fn list_visible_providers(&self) -> Vec<FederatedProviderView>;
/// Compatibility accessor for existing provider implementations.
/// Login redirect consumers should use [`Self::redirect_policy`].
fn provider_config(&self, _provider_id: &str) -> Option<&OidcProviderConfig> {
None
}
fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
self.provider_config(provider_id).map(|config| FederatedRedirectPolicy {
redirect_uri: config.redirect_uri.clone(),
allow_request_origin: config.redirect_uri_dynamic,
})
}
async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String>;
async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<FederatedCodeExchange>;
async fn verify_web_identity_token(&self, jwt: &str) -> Result<FederatedAuthorization>;
async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String>;
async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>>;
fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy>;
}
#[cfg(test)]
-32
View File
@@ -1,32 +0,0 @@
// 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.
use super::FederatedIdentityProvider;
use std::sync::Arc;
/// Compatibility facade for integrations built against the original provider API.
/// Production standard OIDC startup composes the adapter, mapper, and service directly.
pub struct FederatedIdentityRegistry {
standard_oidc: Arc<dyn FederatedIdentityProvider>,
}
impl FederatedIdentityRegistry {
pub fn new(standard_oidc: Arc<dyn FederatedIdentityProvider>) -> Self {
Self { standard_oidc }
}
pub(crate) fn standard_oidc_arc(&self) -> Arc<dyn FederatedIdentityProvider> {
Arc::clone(&self.standard_oidc)
}
}
+105 -217
View File
@@ -13,11 +13,10 @@
// limitations under the License.
use super::{
CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedIdentityProvider, FederatedIdentityRegistry,
FederatedLoginSession, FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding,
FederatedSessionTransaction, FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
CoreFederatedAuthorizationMapper, FederatedAuthorization, FederatedLoginSession, FederatedProviderQuery,
FederatedProviderView, FederatedRedirectPolicy, FederatedSession, FederatedSessionBinding, FederatedSessionTransaction,
FederationError, OpaqueLogoutContinuation, Result, StandardOidcAuthentication,
};
use crate::oidc::OidcProviderConfig;
use std::sync::Arc;
const DEFAULT_OIDC_PROVIDER_ID: &str = "default";
@@ -32,11 +31,7 @@ fn sorted_provider_views(mut providers: Vec<FederatedProviderView>) -> Vec<Feder
}
pub struct FederatedIdentityService {
provider: Arc<dyn FederatedIdentityProvider>,
standard_oidc: Option<StandardOidcRuntime>,
}
struct StandardOidcRuntime {
provider_query: Arc<dyn FederatedProviderQuery>,
authentication: Arc<dyn StandardOidcAuthentication>,
mapper: CoreFederatedAuthorizationMapper,
}
@@ -49,91 +44,57 @@ struct RuntimeCodeExchange {
impl FederatedIdentityService {
async fn exchange_code(&self, state: &str, code: &str, redirect_uri: &str) -> Result<RuntimeCodeExchange> {
match &self.standard_oidc {
Some(runtime) => {
let exchange = runtime.authentication.exchange_identity(state, code, redirect_uri).await?;
let (identity, redirect_after, logout_continuation) = exchange.into_parts();
Ok(RuntimeCodeExchange {
authorization: runtime.mapper.map(identity),
redirect_after,
logout_continuation,
})
}
None => {
let exchange = self.provider.exchange_code(state, code, redirect_uri).await?;
let (authorization, redirect_after, logout_continuation) = exchange.into_parts();
Ok(RuntimeCodeExchange {
authorization,
redirect_after,
logout_continuation,
})
}
}
let exchange = self.authentication.exchange_identity(state, code, redirect_uri).await?;
let (identity, redirect_after, logout_continuation) = exchange.into_parts();
Ok(RuntimeCodeExchange {
authorization: self.mapper.map(identity),
redirect_after,
logout_continuation,
})
}
async fn verify_identity(&self, jwt: &str) -> Result<FederatedAuthorization> {
match &self.standard_oidc {
Some(runtime) => Ok(runtime.mapper.map(runtime.authentication.verify_identity(jwt).await?)),
None => self.provider.verify_web_identity_token(jwt).await,
}
Ok(self.mapper.map(self.authentication.verify_identity(jwt).await?))
}
async fn create_logout_token(&self, continuation: OpaqueLogoutContinuation) -> Result<String> {
match &self.standard_oidc {
Some(runtime) => runtime.authentication.create_logout_token(continuation).await,
None => {
let (provider, id_token) = continuation.into_parts();
self.provider.create_logout_token(provider.as_str(), &id_token).await
}
}
}
/// Build the compatibility service around the published provider interface.
pub fn new(registry: FederatedIdentityRegistry) -> Self {
Self {
provider: registry.standard_oidc_arc(),
standard_oidc: None,
}
self.authentication.create_logout_token(continuation).await
}
pub(super) fn from_standard_oidc_parts(
provider: Arc<dyn FederatedIdentityProvider>,
provider_query: Arc<dyn FederatedProviderQuery>,
authentication: Arc<dyn StandardOidcAuthentication>,
mapper: CoreFederatedAuthorizationMapper,
) -> Self {
Self {
provider,
standard_oidc: Some(StandardOidcRuntime { authentication, mapper }),
provider_query,
authentication,
mapper,
}
}
pub fn has_providers(&self) -> bool {
self.provider.has_providers()
self.provider_query.has_providers()
}
pub fn list_providers(&self) -> Vec<FederatedProviderView> {
let providers = self.provider.list_providers();
let providers = self.provider_query.list_providers();
sorted_provider_views(providers)
}
pub fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
let providers = self.provider.list_visible_providers();
let providers = self.provider_query.list_visible_providers();
sorted_provider_views(providers)
}
pub fn redirect_policy(&self, provider_id: &str) -> Option<FederatedRedirectPolicy> {
self.provider.redirect_policy(provider_id)
}
/// Compatibility accessor for existing federation consumers.
/// Login redirects use [`Self::redirect_policy`], while OIDC-specific
/// configuration consumers use the OIDC configuration query.
pub fn get_provider_config(&self, provider_id: &str) -> Option<&OidcProviderConfig> {
self.provider.provider_config(provider_id)
self.provider_query.redirect_policy(provider_id)
}
pub async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
self.provider.authorize_url(provider_id, redirect_uri, redirect_after).await
self.authentication
.authorize_url(provider_id, redirect_uri, redirect_after)
.await
}
pub async fn complete_authorization_code(
@@ -205,7 +166,9 @@ impl FederatedIdentityService {
}
pub async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
self.provider.build_logout_url(logout_token, post_logout_redirect_uri).await
self.authentication
.build_logout_url(logout_token, post_logout_redirect_uri)
.await
}
}
@@ -213,9 +176,8 @@ impl FederatedIdentityService {
mod tests {
use super::*;
use crate::federation::{
FederatedAuthorization, FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedCodeExchange,
FederatedIdentityProvider, FederatedProviderRef, FederatedSessionBindingError, VerifiedFederatedCodeExchange,
VerifiedFederatedIdentity,
FederatedAuthorizationRule, FederatedAuthorizationRules, FederatedClaims, FederatedProviderQuery, FederatedProviderRef,
FederatedSessionBindingError, VerifiedFederatedCodeExchange, VerifiedFederatedIdentity,
};
use rustfs_credentials::Credentials;
use std::sync::{Arc, Mutex};
@@ -235,6 +197,8 @@ mod tests {
web_provider_id: &'static str,
failure: ProviderFailure,
events: Arc<Mutex<Vec<&'static str>>>,
authorize_calls: Mutex<Vec<(String, String, Option<String>)>>,
logout_url_calls: Mutex<Vec<(String, String)>>,
expected_logout: (&'static str, &'static str),
listed_provider_ids: Vec<&'static str>,
visible_provider_ids: Vec<&'static str>,
@@ -249,6 +213,8 @@ mod tests {
web_provider_id: DEFAULT_OIDC_PROVIDER_ID,
failure: ProviderFailure::None,
events,
authorize_calls: Mutex::new(Vec::new()),
logout_url_calls: Mutex::new(Vec::new()),
expected_logout: (DEFAULT_OIDC_PROVIDER_ID, "id-token"),
listed_provider_ids: Vec::new(),
visible_provider_ids: Vec::new(),
@@ -259,32 +225,6 @@ mod tests {
self.events.lock().expect("event log should not be poisoned").push(event);
}
fn authorization(&self, provider_id: &str) -> FederatedAuthorization {
FederatedAuthorization {
provider_id: provider_id.to_string(),
claims: FederatedClaims {
sub: "subject".to_string(),
email: String::new(),
username: "user".to_string(),
groups: vec!["source-group".to_string()],
raw: Default::default(),
},
policies: if self.with_policy {
vec!["readwrite".to_string()]
} else {
Vec::new()
},
group_claim_policies: Vec::new(),
groups: if self.with_group {
vec!["developers".to_string()]
} else {
Vec::new()
},
roles_claim_key: None,
roles: Vec::new(),
}
}
fn identity(&self, provider_id: &str) -> VerifiedFederatedIdentity {
VerifiedFederatedIdentity::from_claims(
FederatedProviderRef::new(provider_id.to_string()),
@@ -313,8 +253,7 @@ mod tests {
.collect()
}
#[async_trait::async_trait]
impl FederatedIdentityProvider for TestProvider {
impl FederatedProviderQuery for TestProvider {
fn has_providers(&self) -> bool {
true
}
@@ -333,52 +272,18 @@ mod tests {
allow_request_origin: true,
})
}
async fn authorize_url(
&self,
_provider_id: &str,
_redirect_uri: &str,
_redirect_after: Option<String>,
) -> Result<String> {
Ok("https://identity.example/authorize".to_string())
}
async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> Result<FederatedCodeExchange> {
self.record("exchange");
if self.failure == ProviderFailure::Exchange {
return Err(FederationError::CodeExchange("exchange failed".to_string()));
}
Ok(FederatedCodeExchange {
authorization: self.authorization(self.browser_provider_id),
redirect_after: Some("/browser".to_string()),
id_token: "id-token".to_string(),
})
}
async fn verify_web_identity_token(&self, _jwt: &str) -> Result<FederatedAuthorization> {
self.record("verify");
if self.failure == ProviderFailure::Verification {
return Err(FederationError::TokenVerification("verification failed".to_string()));
}
Ok(self.authorization(self.web_provider_id))
}
async fn create_logout_token(&self, provider_id: &str, id_token: &str) -> Result<String> {
self.record("logout");
assert_eq!((provider_id, id_token), self.expected_logout);
if self.failure == ProviderFailure::Logout {
return Err(FederationError::Logout("logout failed".to_string()));
}
Ok("logout-token".to_string())
}
async fn build_logout_url(&self, _logout_token: &str, _post_logout_redirect_uri: &str) -> Result<Option<String>> {
Ok(Some("https://identity.example/logout".to_string()))
}
}
#[async_trait::async_trait]
impl StandardOidcAuthentication for TestProvider {
async fn authorize_url(&self, provider_id: &str, redirect_uri: &str, redirect_after: Option<String>) -> Result<String> {
self.authorize_calls
.lock()
.expect("authorize call log should not be poisoned")
.push((provider_id.to_string(), redirect_uri.to_string(), redirect_after));
Ok("https://identity.example/authorize".to_string())
}
async fn exchange_identity(
&self,
_state: &str,
@@ -414,6 +319,14 @@ mod tests {
}
Ok("logout-token".to_string())
}
async fn build_logout_url(&self, logout_token: &str, post_logout_redirect_uri: &str) -> Result<Option<String>> {
self.logout_url_calls
.lock()
.expect("logout URL call log should not be poisoned")
.push((logout_token.to_string(), post_logout_redirect_uri.to_string()));
Ok(Some("https://identity.example/logout".to_string()))
}
}
fn standard_service(provider: Arc<TestProvider>) -> FederatedIdentityService {
@@ -432,14 +345,12 @@ mod tests {
)
}));
let authentication: Arc<dyn StandardOidcAuthentication> = provider.clone();
let provider: Arc<dyn FederatedIdentityProvider> = provider;
FederatedIdentityService {
provider,
standard_oidc: Some(StandardOidcRuntime {
authentication,
mapper: CoreFederatedAuthorizationMapper::new(rules),
}),
}
let provider_query: Arc<dyn FederatedProviderQuery> = provider;
FederatedIdentityService::from_standard_oidc_parts(
provider_query,
authentication,
CoreFederatedAuthorizationMapper::new(rules),
)
}
struct RecordingBinding {
@@ -492,7 +403,7 @@ mod tests {
let mut provider = TestProvider::new(events);
provider.listed_provider_ids = vec!["zeta", "hidden", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
provider.visible_provider_ids = vec!["zeta", DEFAULT_OIDC_PROVIDER_ID, "alpha"];
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
let service = standard_service(Arc::new(provider));
assert_eq!(
service
@@ -519,6 +430,48 @@ mod tests {
);
}
#[tokio::test]
async fn authorization_and_logout_urls_preserve_service_arguments() {
let provider = Arc::new(TestProvider::new(Arc::new(Mutex::new(Vec::new()))));
let service = standard_service(provider.clone());
let authorize_url = service
.authorize_url(
"corp",
"https://console.example.com/oauth/callback",
Some("/browser/path?tab=objects".to_string()),
)
.await
.expect("authorization URL should be returned");
let logout_url = service
.build_logout_url("opaque-logout-token", "https://console.example.com/signed-out")
.await
.expect("logout URL should be returned");
assert_eq!(authorize_url, "https://identity.example/authorize");
assert_eq!(logout_url.as_deref(), Some("https://identity.example/logout"));
assert_eq!(
provider
.authorize_calls
.lock()
.expect("authorize call log should not be poisoned")
.as_slice(),
[(
"corp".to_string(),
"https://console.example.com/oauth/callback".to_string(),
Some("/browser/path?tab=objects".to_string()),
)]
);
assert_eq!(
provider
.logout_url_calls
.lock()
.expect("logout URL call log should not be poisoned")
.as_slice(),
[("opaque-logout-token".to_string(), "https://console.example.com/signed-out".to_string(),)]
);
}
#[tokio::test]
async fn callback_and_web_identity_preserve_provider_and_transaction_boundaries() {
let events = Arc::new(Mutex::new(Vec::new()));
@@ -526,9 +479,8 @@ mod tests {
provider.browser_provider_id = "corp";
provider.web_provider_id = "partner";
provider.expected_logout = ("corp", "id-token");
let provider = Arc::new(provider);
let binding = Arc::new(RecordingBinding::new(events.clone()));
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
let service = standard_service(Arc::new(provider));
let login = service
.complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -604,76 +556,13 @@ mod tests {
}
#[tokio::test]
async fn standard_runtime_checks_authorization_before_binding() {
async fn web_identity_without_policy_or_group_is_not_bound() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut provider = TestProvider::new(events.clone());
provider.with_policy = false;
let binding = Arc::new(RecordingBinding::new(events.clone()));
let service = standard_service(Arc::new(provider));
let error = service
.assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
.await
.expect_err("standard web identity requires mapped authorization");
assert!(matches!(error, FederationError::NoAuthorizationContext));
assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify"]);
}
#[tokio::test]
async fn standard_runtime_preserves_browser_failure_order() {
for (provider_failure, binding_failure, expected_events) in [
(ProviderFailure::Exchange, false, vec!["exchange"]),
(ProviderFailure::None, true, vec!["exchange", "bind"]),
(ProviderFailure::Logout, false, vec!["exchange", "bind", "logout"]),
] {
let events = Arc::new(Mutex::new(Vec::new()));
let mut provider = TestProvider::new(events.clone());
provider.failure = provider_failure;
let mut binding = RecordingBinding::new(events.clone());
binding.fail = binding_failure;
let service = standard_service(Arc::new(provider));
service
.complete_authorization_code("state", "code", "https://console.example/callback", 3600, &binding)
.await
.expect_err("the configured standard callback failure should be returned");
assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
}
}
#[tokio::test]
async fn standard_runtime_preserves_web_identity_failure_order() {
for (provider_failure, binding_failure, expected_events) in [
(ProviderFailure::Verification, false, vec!["verify"]),
(ProviderFailure::None, true, vec!["verify", "bind"]),
] {
let events = Arc::new(Mutex::new(Vec::new()));
let mut provider = TestProvider::new(events.clone());
provider.failure = provider_failure;
let mut binding = RecordingBinding::new(events.clone());
binding.fail = binding_failure;
let service = standard_service(Arc::new(provider));
service
.assume_role_with_web_identity("jwt", 3600, None, &binding)
.await
.expect_err("the configured standard web identity failure should be returned");
assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), expected_events);
}
}
#[tokio::test]
async fn web_identity_without_policy_or_group_is_not_bound() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut provider = TestProvider::new(events.clone());
provider.with_policy = false;
let provider = Arc::new(provider);
let binding = Arc::new(RecordingBinding::new(events.clone()));
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
let error = service
.assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
.await
@@ -689,16 +578,15 @@ mod tests {
let mut provider = TestProvider::new(events.clone());
provider.with_policy = false;
provider.with_group = true;
let provider = Arc::new(provider);
let binding = Arc::new(RecordingBinding::new(events.clone()));
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(provider));
let service = standard_service(Arc::new(provider));
let session = service
.assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
.await
.expect("a mapped group is an authorization context");
assert!(session.authorization.policies.is_empty());
assert_eq!(session.authorization.policies, ["developers"]);
assert_eq!(session.authorization.groups, ["developers"]);
assert_eq!(events.lock().expect("event log should not be poisoned").as_slice(), ["verify", "bind"]);
}
@@ -716,7 +604,7 @@ mod tests {
let mut binding = RecordingBinding::new(events.clone());
binding.fail = binding_failure;
let binding = Arc::new(binding);
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
let service = standard_service(Arc::new(provider));
let error = service
.complete_authorization_code("state", "code", "https://console.example/callback", 3600, binding.as_ref())
@@ -755,7 +643,7 @@ mod tests {
let mut binding = RecordingBinding::new(events.clone());
binding.fail = binding_failure;
let binding = Arc::new(binding);
let service = FederatedIdentityService::new(FederatedIdentityRegistry::new(Arc::new(provider)));
let service = standard_service(Arc::new(provider));
let error = service
.assume_role_with_web_identity("jwt", 3600, None, binding.as_ref())
+30 -42
View File
@@ -3345,27 +3345,23 @@ mod tests {
#[test]
fn test_extract_jwt_claims_basic() {
let user_identity = UserIdentity {
version: 1,
credentials: Credentials {
access_key: "test-access-key".to_string(),
secret_key: "test-secret-key".to_string(),
session_token: "invalid-token".to_string(), // Invalid token for testing error handling
expiration: None,
status: "enabled".to_string(),
parent_user: "".to_string(),
groups: None,
claims: Some({
let mut claims = HashMap::new();
claims.insert("sub".to_string(), json!("test-user"));
claims.insert("aud".to_string(), json!("test-audience"));
claims
}),
name: None,
description: None,
},
update_at: Some(OffsetDateTime::now_utc()),
};
let user_identity = UserIdentity::new(Credentials {
access_key: "test-access-key".to_string(),
secret_key: "test-secret-key".to_string(),
session_token: "invalid-token".to_string(), // Invalid token for testing error handling
expiration: None,
status: "enabled".to_string(),
parent_user: "".to_string(),
groups: None,
claims: Some({
let mut claims = HashMap::new();
claims.insert("sub".to_string(), json!("test-user"));
claims.insert("aud".to_string(), json!("test-audience"));
claims
}),
name: None,
description: None,
});
let result = extract_jwt_claims(&user_identity);
// In test environment without proper JWT setup, this should fail
@@ -3374,22 +3370,18 @@ mod tests {
#[test]
fn test_extract_jwt_claims_no_claims() {
let user_identity = UserIdentity {
version: 1,
credentials: Credentials {
access_key: "test-access-key".to_string(),
secret_key: "test-secret-key".to_string(),
session_token: "".to_string(), // Empty token
expiration: None,
status: "enabled".to_string(),
parent_user: "".to_string(),
groups: None,
claims: None,
name: None,
description: None,
},
update_at: Some(OffsetDateTime::now_utc()),
};
let user_identity = UserIdentity::new(Credentials {
access_key: "test-access-key".to_string(),
secret_key: "test-secret-key".to_string(),
session_token: "".to_string(), // Empty token
expiration: None,
status: "enabled".to_string(),
parent_user: "".to_string(),
groups: None,
claims: None,
name: None,
description: None,
});
let result = extract_jwt_claims(&user_identity);
// Should fail with empty session token
@@ -3448,11 +3440,7 @@ mod tests {
description: None,
};
let user_identity = UserIdentity {
version: 1,
credentials,
update_at: Some(OffsetDateTime::now_utc()),
};
let user_identity = UserIdentity::new(credentials);
// Test basic structure
assert_eq!(user_identity.version, 1);
+7 -5
View File
@@ -2354,8 +2354,9 @@ fn extract_canonical_group_values(
groups
}
#[cfg(test)]
pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
#[cfg(any(test, feature = "test-util"))]
#[doc(hidden)]
pub fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
let configs = configs
.into_iter()
.map(test_sourced_config)
@@ -2369,7 +2370,7 @@ pub(crate) fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
}
}
#[cfg(test)]
#[cfg(any(test, feature = "test-util"))]
fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig {
SourcedOidcProviderConfig {
config,
@@ -2377,8 +2378,9 @@ fn test_sourced_config(config: OidcProviderConfig) -> SourcedOidcProviderConfig
}
}
#[cfg(test)]
pub(crate) fn test_config(id: &str) -> OidcProviderConfig {
#[cfg(any(test, feature = "test-util"))]
#[doc(hidden)]
pub fn test_config(id: &str) -> OidcProviderConfig {
OidcProviderConfig {
id: id.to_string(),
enabled: true,
+1
View File
@@ -91,6 +91,7 @@ pub(crate) mod crate_boundary {
pub(crate) mod object_store {
pub(crate) use super::storage_contracts::{HTTPPreconditions, ListOperations, ObjectInfoOrErr, ObjectOperations};
pub(crate) use rustfs_ecstore::api::config::com::read_config_preserve_empty_with_metadata_opts;
}
pub(crate) mod runtime {
+10 -3
View File
@@ -42,6 +42,7 @@ use tracing::{debug, error, warn};
use crate::storage_api::object_store::{
HTTPPreconditions, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, ObjectOperations,
read_config_preserve_empty_with_metadata_opts,
};
pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam"));
@@ -414,9 +415,14 @@ impl ObjectStore {
&self,
path: impl AsRef<str> + Send,
mode: LoadMode,
preserve_empty: bool,
) -> Result<(Vec<u8>, IamObjectInfo)> {
let path_ref = path.as_ref();
let (data, obj) = read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?;
let (data, obj) = if preserve_empty {
read_config_preserve_empty_with_metadata_opts(self.object_api.clone(), path_ref, &mode.read_opts()).await?
} else {
read_iam_config_with_metadata(self.object_api.clone(), path_ref, &mode.read_opts()).await?
};
let outcome = Self::decrypt_data_with_source(&data)?;
self.maybe_schedule_lazy_rewrite(path_ref, &outcome, &obj);
@@ -614,8 +620,9 @@ impl ObjectStore {
/// Parameterized core of [`Store::load_user_identity`].
async fn load_user_identity_with(&self, name: &str, user_type: UserType, mode: LoadMode) -> Result<UserIdentity> {
let mut u: UserIdentity = self
.load_iam_config_with(get_user_identity_path(name, user_type), mode)
.load_iamconfig_bytes_with_metadata(get_user_identity_path(name, user_type), mode, true)
.await
.and_then(|(data, _)| serde_json::from_slice(&data).map_err(Error::from))
.map_err(|err| {
if is_err_config_not_found(&err) {
debug!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing");
@@ -695,7 +702,7 @@ impl ObjectStore {
/// Parameterized core of [`Store::load_policy`].
async fn load_policy_with(&self, name: &str, mode: LoadMode) -> Result<PolicyDoc> {
let (data, obj) = self
.load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode)
.load_iamconfig_bytes_with_metadata(get_policy_doc_path(name), mode, false)
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
+8 -20
View File
@@ -2667,11 +2667,7 @@ mod tests {
name: None,
description: None,
};
let parent_identity = UserIdentity {
version: 1,
credentials: creds,
update_at: Some(OffsetDateTime::now_utc()),
};
let parent_identity = UserIdentity::new(creds);
let mut users = HashMap::new();
users.insert(PARENT_USER.to_string(), parent_identity);
@@ -2703,11 +2699,7 @@ mod tests {
name: None,
description: None,
};
let parent_identity = UserIdentity {
version: 1,
credentials: creds,
update_at: Some(OffsetDateTime::now_utc()),
};
let parent_identity = UserIdentity::new(creds);
let mut users = HashMap::new();
users.insert(PARENT_USER.to_string(), parent_identity);
@@ -2820,16 +2812,12 @@ mod tests {
async fn group_writes_stamped_before_the_cache_load_time_still_publish() {
let iam_sys = test_iam_sys().await;
let member = "group-stamp-member";
let identity = UserIdentity {
version: 1,
credentials: Credentials {
access_key: member.to_string(),
secret_key: "longenoughsecret".to_string(),
status: "on".to_string(),
..Default::default()
},
update_at: Some(OffsetDateTime::now_utc()),
};
let identity = UserIdentity::new(Credentials {
access_key: member.to_string(),
secret_key: "longenoughsecret".to_string(),
status: "on".to_string(),
..Default::default()
});
iam_sys.store.cache.with_write_lock(|cache| {
cache.add_or_update_user(member, &identity, OffsetDateTime::now_utc());
// The startup load publishes every entity with the load time.
@@ -22,7 +22,11 @@
#[allow(unused_imports)]
pub(crate) mod fixture {
pub(crate) use rustfs_ecstore::api::bucket::migration::try_migrate_iam_config;
pub(crate) use rustfs_ecstore::api::config::com::save_config;
pub(crate) use rustfs_ecstore::api::global::set_object_store_resolver;
pub(crate) use rustfs_ecstore::api::layout::SetupType;
pub(crate) use rustfs_ecstore::api::runtime::{InstanceContext, first_cluster_node_is_local, object_store_handle};
pub(crate) use rustfs_ecstore::api::storage::ECStore;
// `update_erasure_type` is a write-side global facade entry. Its use is
// restricted to reviewed storage_api boundaries; this test-only module is
+458 -4
View File
@@ -33,12 +33,21 @@
mod ecstore_test_compat;
use ecstore_test_compat::fixture::{SetupType, update_erasure_type};
use ecstore_test_compat::fixture::{
ECStore, InstanceContext, SetupType, first_cluster_node_is_local, object_store_handle, save_config,
set_object_store_resolver, update_erasure_type,
};
use rustfs_credentials::{Credentials, IAM_POLICY_CLAIM_NAME_SA};
use rustfs_iam::cache::Cache;
use rustfs_iam::store::object::ObjectStore;
use rustfs_iam::store::{GroupInfo, Store};
use rustfs_iam::error::Error as IamError;
use rustfs_iam::manager::{IamCache, IamState};
use rustfs_iam::store::object::{IAM_CONFIG_POLICIES_PREFIX, IAM_CONFIG_USERS_PREFIX, ObjectStore};
use rustfs_iam::store::{GroupInfo, Store, UserType};
use rustfs_policy::auth::UserIdentity;
use serial_test::serial;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU8, AtomicU64};
use std::sync::{Arc, Mutex, OnceLock, Weak};
const TEST_GROUP: &str = "seq-restart-group";
const TEST_MEMBERS: [&str; 2] = ["alice", "bob"];
@@ -69,7 +78,8 @@ async fn load_all_bypasses_namespace_lock_quorum() {
.build()
.await
.ecstore;
let store = ObjectStore::new(ecstore);
let _route_guard = TestStoreRouteGuard::new(&ecstore);
let store = ObjectStore::new(ecstore.clone());
// Seed IAM data while namespace locks still work (single-node mode).
store
@@ -128,3 +138,447 @@ async fn load_all_bypasses_namespace_lock_quorum() {
})
.await;
}
const IDENTITY_PARENT: &str = "fixture-parent";
const IDENTITY_CHILD: &str = "fixture-service-child";
static TEST_STORE_ROUTE: Mutex<Option<Weak<ECStore>>> = Mutex::new(None);
static TEST_STORE_RESOLVER: OnceLock<()> = OnceLock::new();
struct TestStoreRouteGuard;
impl TestStoreRouteGuard {
fn new(ecstore: &Arc<ECStore>) -> Self {
TEST_STORE_RESOLVER.get_or_init(|| {
assert!(
set_object_store_resolver(Arc::new(|| {
TEST_STORE_ROUTE
.lock()
.expect("lock test store route")
.as_ref()
.and_then(Weak::upgrade)
})),
"test binary must own its resolver registration"
);
});
let mut route = TEST_STORE_ROUTE.lock().expect("lock test store route");
assert!(route.is_none(), "another test store route is active");
*route = Some(Arc::downgrade(ecstore));
Self
}
}
impl Drop for TestStoreRouteGuard {
fn drop(&mut self) {
*TEST_STORE_ROUTE.lock().expect("lock test store route for cleanup") = None;
}
}
struct IdentityStoreFixture {
previous_store: Option<Arc<ECStore>>,
ecstore: Arc<ECStore>,
manager: IamCache<ObjectStore>,
_route_guard: TestStoreRouteGuard,
_temp_dir: tempfile::TempDir,
}
impl IdentityStoreFixture {
async fn new() -> Self {
let temp_dir = tempfile::TempDir::with_prefix("rustfs_iam_identity_fixture_").expect("create identity fixture directory");
let ctx = Arc::new(InstanceContext::new());
let env = rustfs_test_utils::TestECStoreEnv::builder()
.base_dir(temp_dir.path())
.init_bucket_metadata(false)
.instance_ctx(ctx.clone())
.build()
.await;
ctx.set_endpoints(env.endpoint_pools.clone());
let ecstore = env.ecstore;
let previous_store = object_store_handle();
let route_guard = TestStoreRouteGuard::new(&ecstore);
let selected = object_store_handle().expect("fixture route selects its ECStore");
assert!(Arc::ptr_eq(&selected, &ecstore), "fixture route must select the isolated ECStore");
drop(selected);
assert!(first_cluster_node_is_local().await, "isolated fixture has a local first endpoint");
temp_env::async_with_vars([("RUSTFS_SKIP_BACKGROUND_TASK", Some("1"))], rustfs_iam::build_iam_sys(ecstore.clone()))
.await
.expect("initialize IAM metadata through production owner");
let store = ObjectStore::new(ecstore.clone());
let parent = UserIdentity::new(Credentials {
access_key: IDENTITY_PARENT.to_string(),
secret_key: "fixture-parent-secret".to_string(),
expiration: None,
session_token: String::new(),
status: "on".to_string(),
..Default::default()
});
let child = UserIdentity::new(Credentials {
access_key: IDENTITY_CHILD.to_string(),
secret_key: "fixture-child-secret".to_string(),
expiration: None,
session_token: String::new(),
status: "on".to_string(),
parent_user: IDENTITY_PARENT.to_string(),
claims: Some(HashMap::from([(
IAM_POLICY_CLAIM_NAME_SA.to_string(),
serde_json::json!("inherited-policy"),
)])),
..Default::default()
});
assert!(child.credentials.is_service_account(), "child fixture must be a service account");
store
.save_user_identity(IDENTITY_PARENT, UserType::Reg, parent, None)
.await
.expect("persist parent identity");
store
.save_user_identity(IDENTITY_CHILD, UserType::Svc, child, None)
.await
.expect("persist service child identity");
let manager = IamCache {
cache: Cache::default(),
api: store,
state: Arc::new(AtomicU8::new(IamState::Ready as u8)),
loading: Arc::new(AtomicBool::new(false)),
roles: HashMap::new(),
send_chan: tokio::sync::mpsc::channel::<i64>(1).0,
last_timestamp: AtomicI64::new(0),
sync_failures: AtomicU64::new(0),
sync_successes: AtomicU64::new(0),
last_sync_duration_millis: AtomicU64::new(0),
};
manager
.load_user(IDENTITY_PARENT)
.await
.expect("load persisted parent into cache");
manager
.load_user(IDENTITY_CHILD)
.await
.expect("load persisted service child into cache");
Self {
previous_store,
ecstore,
manager,
_route_guard: route_guard,
_temp_dir: temp_dir,
}
}
async fn assert_parent_and_child(&self) {
let parent = self
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect("read parent from real storage");
assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
assert_eq!(parent.credentials.parent_user, "");
let child = self
.manager
.api
.load_user_identity(IDENTITY_CHILD, UserType::Svc)
.await
.expect("read child from real storage");
assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
assert!(child.credentials.is_service_account());
let cached_parent = self.manager.get_user(IDENTITY_PARENT).await.expect("parent remains in cache");
assert_eq!(cached_parent.credentials.access_key, IDENTITY_PARENT);
assert_eq!(cached_parent.credentials.parent_user, "");
let cached_child = self
.manager
.get_user(IDENTITY_CHILD)
.await
.expect("service child remains in cache");
assert_eq!(cached_child.credentials.access_key, IDENTITY_CHILD);
assert_eq!(cached_child.credentials.parent_user, IDENTITY_PARENT);
assert!(cached_child.credentials.is_service_account());
}
async fn overwrite_parent_bytes(&self, bytes: &[u8]) {
let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
save_config(self.ecstore.clone(), &path, bytes.to_vec())
.await
.expect("overwrite real parent identity bytes");
}
async fn assert_child_storage_and_identity_cache(&self) {
let child = self
.manager
.api
.load_user_identity(IDENTITY_CHILD, UserType::Svc)
.await
.expect("notification must preserve persisted service child");
assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
assert!(child.credentials.is_service_account());
let parent = self
.manager
.get_user(IDENTITY_PARENT)
.await
.expect("failed notification preserves cached parent");
assert_eq!(parent.credentials.access_key, IDENTITY_PARENT);
assert_eq!(parent.credentials.parent_user, "");
let child = self
.manager
.get_user(IDENTITY_CHILD)
.await
.expect("failed notification preserves cached child");
assert_eq!(child.credentials.access_key, IDENTITY_CHILD);
assert_eq!(child.credentials.parent_user, IDENTITY_PARENT);
assert!(child.credentials.is_service_account());
}
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_store_fixture_roundtrips_parent_and_service_child() {
let fixture = IdentityStoreFixture::new().await;
let before = fixture.previous_store.clone();
let selected = object_store_handle().expect("fixture route selects its ECStore");
assert!(Arc::ptr_eq(&selected, &fixture.ecstore));
drop(selected);
fixture.assert_parent_and_child().await;
drop(fixture);
assert!(TEST_STORE_ROUTE.lock().expect("inspect cleared test store route").is_none());
match (before, object_store_handle()) {
(None, None) => {}
(Some(before), Some(after)) => assert!(Arc::ptr_eq(&before, &after), "fixture drop restores prior store identity"),
_ => panic!("fixture drop must restore the prior store option"),
}
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_notification_valid_parent_preserves_child_storage_and_cache() {
let fixture = IdentityStoreFixture::new().await;
fixture.assert_parent_and_child().await;
fixture
.manager
.user_notification_handler(IDENTITY_PARENT, UserType::Reg)
.await
.expect("refresh valid parent notification");
fixture.assert_parent_and_child().await;
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_store_fixture_raw_valid_overwrite_is_observed() {
let fixture = IdentityStoreFixture::new().await;
let parent = fixture
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect("load valid parent before raw overwrite");
let mut raw = serde_json::to_value(parent).expect("serialize valid parent fixture");
raw["credentials"]["name"] = serde_json::json!("updated-name");
let path = format!("{}{IDENTITY_PARENT}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
save_config(
fixture.ecstore.clone(),
&path,
serde_json::to_vec(&raw).expect("encode raw parent fixture"),
)
.await
.expect("overwrite parent through real config storage");
fixture
.manager
.load_user(IDENTITY_PARENT)
.await
.expect("reload parent after raw overwrite");
let stored = fixture
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect("read overwritten parent from storage");
assert_eq!(stored.credentials.name.as_deref(), Some("updated-name"));
let cached = fixture
.manager
.get_user(IDENTITY_PARENT)
.await
.expect("read overwritten parent from cache");
assert_eq!(cached.credentials.name.as_deref(), Some("updated-name"));
fixture.assert_parent_and_child().await;
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_corrupt_reads_preserve_crypto_error() {
let fixture = IdentityStoreFixture::new().await;
fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
let locked = fixture
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect_err("corrupt stored identity must fail locked read");
assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
let mut users = HashMap::new();
let no_lock = fixture
.manager
.api
.load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
.await
.expect_err("corrupt stored identity must fail no-lock read");
assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
assert!(users.is_empty(), "failed no-lock read must not populate users");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_corrupt_notification_preserves_child_storage_and_cache() {
let fixture = IdentityStoreFixture::new().await;
fixture.assert_parent_and_child().await;
fixture.overwrite_parent_bytes(&[0xff, 0xfe, 0xfd]).await;
let notification = fixture
.manager
.user_notification_handler(IDENTITY_PARENT, UserType::Reg)
.await;
fixture.assert_child_storage_and_identity_cache().await;
let error = notification.expect_err("corrupt parent notification must return an error");
assert!(
matches!(error, IamError::CryptoError(_)),
"preserve notification crypto error, got {error:?}"
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_invalid_schema_preserves_decode_error() {
let fixture = IdentityStoreFixture::new().await;
fixture.assert_parent_and_child().await;
fixture
.overwrite_parent_bytes(br#"{"version":"not-an-integer","credentials":{}}"#)
.await;
let error = fixture
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect_err("invalid identity field type must fail decode");
assert_identity_data_error(error);
let mut users = HashMap::new();
let error = fixture
.manager
.api
.load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
.await
.expect_err("invalid identity field type must fail no-lock decode");
assert_identity_data_error(error);
assert!(users.is_empty(), "failed decode must not populate users");
let notification = fixture
.manager
.user_notification_handler(IDENTITY_PARENT, UserType::Reg)
.await;
fixture.assert_child_storage_and_identity_cache().await;
assert_identity_data_error(notification.expect_err("invalid parent notification must return decode error"));
}
fn assert_identity_data_error(error: IamError) {
let IamError::Io(inner) = error else {
panic!("identity decode must preserve Io serde error, got {error:?}");
};
let cause = inner
.get_ref()
.and_then(|cause| cause.downcast_ref::<serde_json::Error>())
.expect("identity decode error retains serde source");
assert!(cause.is_data(), "identity field type must produce serde Data error");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_physically_missing_is_no_such_user() {
let fixture = IdentityStoreFixture::new().await;
let name = "physically-missing-identity";
let error = fixture
.manager
.api
.load_user_identity(name, UserType::Reg)
.await
.expect_err("missing identity must be absent");
assert!(
matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
"preserve absent name, got {error:?}"
);
let mut users = HashMap::new();
let error = fixture
.manager
.api
.load_user_no_lock(name, UserType::Reg, &mut users)
.await
.expect_err("missing no-lock identity must be absent");
assert!(
matches!(error, IamError::NoSuchUser(ref missing) if missing == name),
"preserve absent name, got {error:?}"
);
assert!(users.is_empty(), "missing read must not populate users");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_empty_reads_preserve_crypto_error() {
let fixture = IdentityStoreFixture::new().await;
fixture.overwrite_parent_bytes(&[]).await;
let locked = fixture
.manager
.api
.load_user_identity(IDENTITY_PARENT, UserType::Reg)
.await
.expect_err("empty stored identity must fail locked read");
assert!(matches!(locked, IamError::CryptoError(_)), "preserve crypto error, got {locked:?}");
let mut users = HashMap::new();
let no_lock = fixture
.manager
.api
.load_user_no_lock(IDENTITY_PARENT, UserType::Reg, &mut users)
.await
.expect_err("empty stored identity must fail no-lock read");
assert!(matches!(no_lock, IamError::CryptoError(_)), "preserve crypto error, got {no_lock:?}");
assert!(users.is_empty(), "failed no-lock read must not populate users");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn identity_empty_notification_preserves_child_storage_and_cache() {
let fixture = IdentityStoreFixture::new().await;
fixture.assert_parent_and_child().await;
fixture.overwrite_parent_bytes(&[]).await;
let notification = fixture
.manager
.user_notification_handler(IDENTITY_PARENT, UserType::Reg)
.await;
fixture.assert_child_storage_and_identity_cache().await;
let error = notification.expect_err("empty parent notification must return an error");
assert!(
matches!(error, IamError::CryptoError(_)),
"preserve notification crypto error, got {error:?}"
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn policy_empty_payload_retains_no_such_policy() {
let fixture = IdentityStoreFixture::new().await;
let name = "empty-policy";
let path = format!("{}{name}/policy.json", IAM_CONFIG_POLICIES_PREFIX.as_str());
save_config(fixture.ecstore.clone(), &path, Vec::new())
.await
.expect("save real empty policy payload");
let locked = fixture
.manager
.api
.load_policy(name)
.await
.err()
.expect("empty policy must retain absent-policy behavior");
assert!(matches!(locked, IamError::NoSuchPolicy), "preserve policy error, got {locked:?}");
let mut policies = HashMap::new();
let no_lock = fixture
.manager
.api
.load_policy_doc_no_lock(name, &mut policies)
.await
.expect_err("empty no-lock policy must retain absent-policy behavior");
assert!(matches!(no_lock, IamError::NoSuchPolicy), "preserve policy error, got {no_lock:?}");
assert!(policies.is_empty(), "failed no-lock policy read must not populate policies");
}
+1
View File
@@ -58,6 +58,7 @@ tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
time = { workspace = true, features = ["serde", "parsing", "formatting", "macros"] }
serde = { workspace = true, features = ["derive", "rc"] }
serde_json = { workspace = true, features = ["raw_value"] }
uuid = { workspace = true, features = ["serde"] }
thiserror.workspace = true
strum = { workspace = true, features = ["derive"] }
rustfs-crypto = { workspace = true }
+142 -1
View File
@@ -22,6 +22,33 @@ use serde_json::json;
use std::collections::HashMap;
use time::OffsetDateTime;
/// A nonnil stored principal identifier; parsing it does not establish identity authority.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(try_from = "uuid::Uuid", into = "uuid::Uuid")]
pub struct PrincipalIncarnation(uuid::Uuid);
#[derive(Debug, thiserror::Error)]
#[error("principal incarnation must not be nil")]
pub struct InvalidPrincipalIncarnation;
impl TryFrom<uuid::Uuid> for PrincipalIncarnation {
type Error = InvalidPrincipalIncarnation;
fn try_from(value: uuid::Uuid) -> Result<Self, Self::Error> {
if value.is_nil() {
Err(InvalidPrincipalIncarnation)
} else {
Ok(Self(value))
}
}
}
impl From<PrincipalIncarnation> for uuid::Uuid {
fn from(value: PrincipalIncarnation) -> Self {
value.0
}
}
#[derive(Serialize, Deserialize, Clone, Debug, Default)]
pub struct UserIdentity {
pub version: i64,
@@ -29,6 +56,8 @@ pub struct UserIdentity {
/// updatedAt (RFC3339), legacy RustFS: update_at. Serialize as updatedAt
#[serde(rename = "updatedAt", alias = "update_at", default, with = "crate::serde_datetime::option")]
pub update_at: Option<OffsetDateTime>,
#[serde(default, rename = "principalIncarnation", skip_serializing_if = "Option::is_none")]
pub principal_incarnation: Option<PrincipalIncarnation>,
}
impl UserIdentity {
@@ -44,6 +73,7 @@ impl UserIdentity {
version: 1,
credentials,
update_at: Some(OffsetDateTime::now_utc()),
principal_incarnation: None,
}
}
@@ -73,13 +103,124 @@ impl From<Credentials> for UserIdentity {
version: 1,
credentials: value,
update_at: Some(OffsetDateTime::now_utc()),
principal_incarnation: None,
}
}
}
#[cfg(test)]
mod tests {
use super::UserIdentity;
use super::{PrincipalIncarnation, UserIdentity};
use rustfs_credentials::Credentials;
use serde_json::json;
use uuid::Uuid;
const INCARNATION: &str = "8f422eba-aab4-4ec3-a7ec-d1717463b520";
#[test]
fn test_user_identity_legacy_fields_absent() {
let mut legacy = json!({"version": 1, "credentials": {"accessKey": "ak", "secretKey": "sk12345678"}});
let identity: UserIdentity = serde_json::from_value(legacy.clone()).expect("decode legacy identity");
assert!(identity.principal_incarnation.is_none());
legacy["principalIncarnation"] = serde_json::Value::Null;
let identity: UserIdentity = serde_json::from_value(legacy).expect("decode null incarnation");
assert!(identity.principal_incarnation.is_none());
}
#[test]
fn test_user_identity_legacy_serialization_shape() {
let identity = UserIdentity::default();
let serialized = serde_json::to_value(&identity).expect("serialize legacy identity");
assert_eq!(serialized, json!({"version": 0, "credentials": identity.credentials, "updatedAt": null}));
assert!(serialized.get("principalIncarnation").is_none());
}
#[test]
fn test_user_identity_constructors_do_not_mint_incarnation() {
let default = UserIdentity::default();
assert!(default.principal_incarnation.is_none());
assert_eq!(default.version, 0);
assert!(default.update_at.is_none());
let credentials = Credentials {
access_key: "ak".to_string(),
secret_key: "sk12345678".to_string(),
..Default::default()
};
let expected_credentials = serde_json::to_value(&credentials).expect("serialize expected credentials");
for identity in [UserIdentity::new(credentials.clone()), UserIdentity::from(credentials)] {
assert!(identity.principal_incarnation.is_none());
assert_eq!(identity.version, 1);
assert!(identity.update_at.is_some());
assert_eq!(
serde_json::to_value(&identity.credentials).expect("serialize constructor credentials"),
expected_credentials
);
assert!(
serde_json::to_value(identity)
.expect("serialize constructor identity")
.get("principalIncarnation")
.is_none()
);
}
}
#[test]
fn test_principal_incarnation_rejects_nil_and_malformed() {
assert!(PrincipalIncarnation::try_from(Uuid::nil()).is_err());
for value in ["00000000-0000-0000-0000-000000000000", "not-a-uuid", ""] {
assert!(serde_json::from_value::<PrincipalIncarnation>(json!(value)).is_err(), "reject {value}");
let identity = json!({"version": 1, "credentials": {}, "principalIncarnation": value});
assert!(serde_json::from_value::<UserIdentity>(identity).is_err(), "identity rejects {value}");
}
}
#[test]
fn test_principal_incarnation_roundtrip_preserves_uuid() {
let uuid = Uuid::parse_str(INCARNATION).expect("parse fixed incarnation");
let incarnation = PrincipalIncarnation::try_from(uuid).expect("accept nonnil incarnation");
assert_eq!(Uuid::from(incarnation).as_bytes(), uuid.as_bytes());
let serialized = serde_json::to_value(incarnation).expect("serialize incarnation");
assert_eq!(serialized, json!(INCARNATION));
assert_eq!(
serde_json::from_value::<PrincipalIncarnation>(serialized).expect("decode incarnation"),
incarnation
);
}
#[test]
fn test_user_identity_live_incarnation_roundtrip() {
let mut identity = UserIdentity::new(Credentials::default());
identity.principal_incarnation = Some(
PrincipalIncarnation::try_from(Uuid::parse_str(INCARNATION).expect("parse fixed incarnation"))
.expect("accept fixed incarnation"),
);
let serialized = serde_json::to_value(&identity).expect("serialize identity with incarnation");
assert_eq!(serialized["principalIncarnation"], INCARNATION);
let encoded = serde_json::to_string(&identity).expect("encode identity with incarnation");
let decoded: UserIdentity = serde_json::from_str(&encoded).expect("decode identity with incarnation");
assert_eq!(decoded.principal_incarnation, identity.principal_incarnation);
assert_eq!(serde_json::to_value(decoded).expect("reserialize identity"), serialized);
}
#[test]
fn test_user_identity_timestamp_aliases_unchanged() {
let mut minio = json!({"version": 1, "credentials": {}, "updatedAt": "2025-03-07T12:00:00Z"});
let encoded = serde_json::to_string(&minio).expect("encode updatedAt fixture");
let identity: UserIdentity = serde_json::from_str(&encoded).expect("decode updatedAt timestamp");
minio["update_at"] = minio
.as_object_mut()
.expect("identity object")
.remove("updatedAt")
.expect("existing timestamp");
let encoded = serde_json::to_string(&minio).expect("encode update_at fixture");
let legacy: UserIdentity = serde_json::from_str(&encoded).expect("decode update_at alias");
assert_eq!(legacy.update_at, identity.update_at);
assert_eq!(legacy.version, 1);
let serialized = serde_json::to_value(legacy).expect("serialize legacy timestamp");
assert_eq!(serialized["updatedAt"], "2025-03-07T12:00:00Z");
assert!(serialized.get("update_at").is_none());
assert!(serialized.get("principalIncarnation").is_none());
}
/// Deserialize UserIdentity from MinIO-style JSON (RFC3339 updatedAt).
#[test]
+89 -19
View File
@@ -252,6 +252,13 @@ impl PolicyVariableResolver for VariableResolver {
}
pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
resolve_aws_variables_with_depth(pattern, resolver, 0).await
}
async fn resolve_aws_variables_with_depth(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
if depth >= 10 {
return vec![pattern.to_string()];
}
let mut results = vec![pattern.to_string()];
let mut changed = true;
@@ -264,7 +271,7 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR
let mut new_results = Vec::new();
for result in &results {
let resolved = resolve_single_pass(result, resolver).await;
let resolved = resolve_single_pass(result, resolver, depth).await;
if resolved.len() > 1 || (resolved.len() == 1 && &resolved[0] != result) {
changed = true;
}
@@ -288,30 +295,31 @@ pub async fn resolve_aws_variables(pattern: &str, resolver: &dyn PolicyVariableR
fn resolve_aws_variables_boxed<'a>(
pattern: &'a str,
resolver: &'a dyn PolicyVariableResolver,
depth: usize,
) -> std::pin::Pin<Box<dyn Future<Output = Vec<String>> + Send + 'a>> {
Box::pin(resolve_aws_variables(pattern, resolver))
Box::pin(resolve_aws_variables_with_depth(pattern, resolver, depth))
}
/// Single pass resolution of variables in a string
async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver) -> Vec<String> {
async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolver, depth: usize) -> Vec<String> {
// Find all ${...} format variables
let mut results = vec![pattern.to_string()];
let mut results = vec![(pattern.to_string(), 0)];
// Process each result string
let mut i = 0;
while i < results.len() {
let mut start = 0;
let mut start = results[i].1;
let mut modified = false;
// Find variables in current string
while let Some(pos) = results[i][start..].find("${") {
while let Some(pos) = results[i].0[start..].find("${") {
let actual_pos = start + pos;
// Find the matching closing brace, taking into account nested braces
let mut brace_count = 1;
let mut end_pos = actual_pos + 2; // Start after "${"
let bytes = results[i].as_bytes();
let bytes = results[i].0.as_bytes();
while end_pos < bytes.len() && brace_count > 0 {
match bytes[end_pos] {
b'{' => brace_count += 1,
@@ -324,19 +332,23 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
}
if brace_count == 0 {
let var_name = &results[i][actual_pos + 2..end_pos];
let var_name = &results[i].0[actual_pos + 2..end_pos];
// Check if this is a nested variable (contains ${...} inside)
if var_name.contains("${") {
// For nested variables like ${${a}-${b}}, we need to resolve the inner variables first
// Then use the resolved result as a new variable to resolve
let resolved_inner = resolve_aws_variables_boxed(var_name, resolver).await;
let resolved_inner = resolve_aws_variables_boxed(var_name, resolver, depth + 1).await;
if resolved_inner.len() == 1 && resolved_inner[0] == var_name {
start = end_pos + 1;
continue;
}
let mut new_results = Vec::new();
for resolved_var_name in resolved_inner {
let prefix = &results[i][..actual_pos];
let suffix = &results[i][end_pos + 1..];
new_results.push(format!("{prefix}{resolved_var_name}{suffix}"));
let prefix = &results[i].0[..actual_pos];
let suffix = &results[i].0[end_pos + 1..];
new_results.push((format!("{prefix}{resolved_var_name}{suffix}"), actual_pos + resolved_var_name.len()));
}
if !new_results.is_empty() {
@@ -354,11 +366,13 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
if !values.is_empty() {
// If there are multiple values, create a new result for each value
let mut new_results = Vec::new();
let prefix = &results[i][..actual_pos];
let suffix = &results[i][end_pos + 1..];
let prefix = &results[i].0[..actual_pos];
let suffix = &results[i].0[end_pos + 1..];
for value in values {
new_results.push(format!("{prefix}{value}{suffix}"));
// Resume after the substitution; a generated placeholder
// belongs to the next bounded pass, not this one.
new_results.push((format!("{prefix}{value}{suffix}"), actual_pos + value.len()));
}
results.splice(i..i + 1, new_results);
@@ -367,9 +381,9 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
} else {
// Variable resolved to empty, just remove the variable placeholder
let mut new_results = Vec::new();
let prefix = &results[i][..actual_pos];
let suffix = &results[i][end_pos + 1..];
new_results.push(format!("{prefix}{suffix}"));
let prefix = &results[i].0[..actual_pos];
let suffix = &results[i].0[end_pos + 1..];
new_results.push((format!("{prefix}{suffix}"), actual_pos));
results.splice(i..i + 1, new_results);
modified = true;
@@ -391,7 +405,7 @@ async fn resolve_single_pass(pattern: &str, resolver: &dyn PolicyVariableResolve
}
}
results
results.into_iter().map(|(value, _)| value).collect()
}
#[cfg(test)]
@@ -399,6 +413,62 @@ mod tests {
use super::*;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
struct CycleResolver {
calls: AtomicUsize,
}
#[async_trait]
impl PolicyVariableResolver for CycleResolver {
async fn resolve(&self, name: &str) -> Option<String> {
// Make the vulnerable implementation terminate so a regression
// reports an assertion failure instead of hanging the test runner.
if self.calls.fetch_add(1, Ordering::Relaxed) >= 100 {
return None;
}
match name {
"aws:AccountId" => Some("$$".to_string()),
"aws:username" => Some("{aws:AccountId}{aws:username}".to_string()),
_ => None,
}
}
fn is_dynamic(&self, _: &str) -> bool {
false
}
}
#[tokio::test]
async fn generated_policy_variable_cycles_are_bounded() {
let resolver = CycleResolver {
calls: AtomicUsize::new(0),
};
let result = resolve_aws_variables("${aws:AccountId}{aws:username}", &resolver).await;
assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
assert_eq!(result, vec!["${aws:AccountId}{aws:username}".to_string()]);
}
#[tokio::test]
async fn a_single_pass_resolves_all_original_variables() {
let mut context = VariableContext::new();
context.username = Some("alice".to_string());
let resolver = VariableResolver::new(context);
let pattern = "${aws:username}".repeat(20);
assert_eq!(resolve_single_pass(&pattern, &resolver, 0).await, vec!["alice".repeat(20)]);
}
#[tokio::test]
async fn nested_policy_variables_have_a_depth_limit() {
let resolver = CycleResolver {
calls: AtomicUsize::new(0),
};
let pattern = format!("{}aws:username{}", "${".repeat(100), "}".repeat(100));
let result = resolve_aws_variables(&pattern, &resolver).await;
assert_eq!(result.len(), 1);
assert!(result[0].contains("${"));
assert!(resolver.calls.load(Ordering::Relaxed) <= 10);
}
#[tokio::test]
async fn test_resolve_aws_variables_with_username() {
+1
View File
@@ -80,6 +80,7 @@ rustfs-tls-runtime.workspace = true
rustfs-utils = { workspace = true, features = ["io", "hash", "compress"] }
serde_json = { workspace = true, features = ["raw_value"] }
md-5 = { workspace = true }
hmac.workspace = true
minlz.workspace = true
tracing.workspace = true
thiserror.workspace = true
+64
View File
@@ -14,8 +14,10 @@
use crate::compress_index::{Index, TryGetIndex};
use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
use hmac::{Hmac, KeyInit, Mac};
use md5::{Digest, Md5};
use pin_project_lite::pin_project;
use sha2::Sha256;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
@@ -29,6 +31,7 @@ pin_project! {
pub finished: bool,
pub checksum: Option<String>,
resolved_etag: Option<String>,
encrypted_etag_mac: Option<Hmac<Sha256>>,
}
}
@@ -40,9 +43,21 @@ impl<R> EtagReader<R> {
finished: false,
checksum,
resolved_etag: None,
encrypted_etag_mac: None,
}
}
/// Protect the content fingerprint with the object's encryption material.
/// Plaintext checksum verification remains the inner reader's responsibility.
pub fn new_encrypted(inner: R, key: [u8; 32], nonce: [u8; 12]) -> std::io::Result<Self> {
let mut mac = <Hmac<Sha256> as KeyInit>::new_from_slice(&key).map_err(std::io::Error::other)?;
mac.update(b"RustFS:encrypted-ETag:v1\0");
mac.update(&nonce);
let mut reader = Self::new(inner, None);
reader.encrypted_etag_mac = Some(mac);
Ok(reader)
}
/// Get the final md5 value (etag) as a hex string, only compute once.
/// Can be called multiple times, always returns the same result after finished.
pub fn get_etag(&mut self) -> String {
@@ -108,6 +123,14 @@ impl<R> EtagResolvable for EtagReader<R> {
true
}
fn try_resolve_etag(&mut self) -> Option<String> {
if let Some(mut mac) = self.encrypted_etag_mac.clone() {
if !self.finished {
return None;
}
mac.update(self.get_etag().as_bytes());
let protected = mac.finalize().into_bytes();
return Some(hex_simd::encode_to_string(&protected[..16], hex_simd::AsciiCase::Lower));
}
// EtagReader provides its own etag, not delegating to inner
if let Some(checksum) = &self.checksum {
Some(checksum.clone())
@@ -148,6 +171,47 @@ mod tests {
use std::io::Cursor;
use tokio::io::{AsyncReadExt, BufReader};
#[tokio::test]
async fn encrypted_etags_hide_plaintext_fingerprints_including_empty_objects() {
for data in [b"".as_slice(), b"Salary 2026: 85000 EUR".as_slice()] {
let md5 = hex_simd::encode_to_string(Md5::digest(data), hex_simd::AsciiCase::Lower);
let mut tags = Vec::new();
for (key, nonce) in [([1; 32], [1; 12]), ([2; 32], [1; 12]), ([1; 32], [2; 12])] {
let mut reader = EtagReader::new_encrypted(Cursor::new(data), key, nonce).expect("create protected ETag reader");
assert!(reader.try_resolve_etag().is_none());
let mut output = Vec::new();
reader.read_to_end(&mut output).await.expect("read unchanged plaintext");
assert_eq!(output, data);
let tag = reader.try_resolve_etag().expect("resolve protected ETag after EOF");
assert_eq!(tag.len(), 32);
assert_ne!(tag, md5);
assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
tags.push(tag);
}
assert_ne!(tags[0], tags[1], "different keys must not link identical content");
assert_ne!(tags[0], tags[2], "different objects under a legacy key must not link identical content");
}
}
#[tokio::test]
async fn encrypted_etag_reader_preserves_plaintext_digest_rejection() {
let plaintext = crate::HashReader::from_stream(
Cursor::new(b"actual plaintext"),
16,
16,
Some("00000000000000000000000000000000".to_owned()),
None,
false,
)
.expect("create checked plaintext reader");
let mut reader = EtagReader::new_encrypted(plaintext, [1; 32], [2; 12]).expect("create protected ETag reader");
let err = reader
.read_to_end(&mut Vec::new())
.await
.expect_err("wrong plaintext MD5 must still be rejected");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
}
#[tokio::test]
async fn test_etag_reader_basic() {
let data = b"hello world";
+5
View File
@@ -136,6 +136,7 @@ pin_project! {
content_sha256: Option<String>,
content_sha256_hasher: Option<Sha256Hasher>,
checksum_on_finish: bool,
pub encrypted_etag: bool,
// Read only at EOF; see the timing contract in `crate::trailer`.
trailer: Option<SharedTrailerSource>,
@@ -184,6 +185,7 @@ impl HashReader {
content_sha256: sha256hex.clone(),
content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
checksum_on_finish: false,
encrypted_etag: false,
trailer: None,
})
}
@@ -224,6 +226,7 @@ impl HashReader {
content_sha256: sha256hex.clone(),
content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
checksum_on_finish: false,
encrypted_etag: false,
trailer: None,
})
}
@@ -294,6 +297,7 @@ impl HashReader {
content_hash,
content_hasher,
checksum_on_finish: false,
encrypted_etag: false,
trailer: existing_hash_reader.get_trailer().cloned(),
})
} else {
@@ -322,6 +326,7 @@ impl HashReader {
content_sha256: sha256hex.clone(),
content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
checksum_on_finish: false,
encrypted_etag: false,
trailer: None,
})
}
+2 -1
View File
@@ -27,9 +27,10 @@ pub(crate) mod fixture {
pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint;
pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints};
pub(crate) use rustfs_ecstore::api::object::{PutObjReader, SelectObjectSnapshot};
pub(crate) use rustfs_ecstore::api::runtime::InstanceContext;
#[cfg(feature = "put-object-commit-barrier")]
pub(crate) use rustfs_ecstore::api::set_disk as ecstore_set_disk;
pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks};
pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks, init_local_disks_with_instance_ctx};
pub(crate) use rustfs_storage_api::{BucketOperations, BucketOptions, MakeBucketOptions, ObjectIO};
#[cfg(test)]
+108 -8
View File
@@ -35,8 +35,9 @@ use std::sync::{Arc, Once};
#[cfg(feature = "put-object-commit-barrier")]
use ecstore_test_compat::fixture::ecstore_set_disk;
use ecstore_test_compat::fixture::{
BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _,
PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
BucketOperations as _, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, MakeBucketOptions,
ObjectIO as _, PoolEndpoints, PutObjReader, SelectObjectSnapshot, init_bucket_metadata_sys, init_local_disks,
init_local_disks_with_instance_ctx,
};
use tokio_util::sync::CancellationToken;
@@ -90,7 +91,7 @@ pub struct TestECStoreEnv {
pub temp_root: PathBuf,
/// The per-disk directories (`disk1`..`diskN`) under `temp_root`.
pub disk_paths: Vec<PathBuf>,
/// The store, bootstrapped exactly like the historical test setups:
/// By default, the store is bootstrapped exactly like the historical test setups:
/// `init_local_disks` + `ECStore::new` on `127.0.0.1:0` (random port keeps
/// nextest's process-per-test parallelism safe).
pub ecstore: Arc<ECStore>,
@@ -155,6 +156,7 @@ pub struct TestECStoreEnvBuilder {
prefix: String,
base_dir: Option<PathBuf>,
init_bucket_metadata: bool,
instance_ctx: Option<Arc<InstanceContext>>,
}
impl Default for TestECStoreEnvBuilder {
@@ -164,6 +166,7 @@ impl Default for TestECStoreEnvBuilder {
prefix: "rustfs_test_utils".to_string(),
base_dir: None,
init_bucket_metadata: true,
instance_ctx: None,
}
}
}
@@ -196,6 +199,12 @@ impl TestECStoreEnvBuilder {
self
}
/// Use a caller-owned context for local disks and store construction.
pub fn instance_ctx(mut self, instance_ctx: Arc<InstanceContext>) -> Self {
self.instance_ctx = Some(instance_ctx);
self
}
/// Build the environment. Panics on any bootstrap failure — this is test
/// scaffolding, and a broken environment must fail the test loudly.
pub async fn build(self) -> TestECStoreEnv {
@@ -237,14 +246,22 @@ impl TestECStoreEnvBuilder {
};
let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]);
init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
// Port 0 keeps ECStore-backed integration binaries parallel-safe under
// nextest: no fixed peer port is ever shared between test processes.
let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().expect("parse test addr");
let ecstore = ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
.await
.expect("build test ECStore");
let ecstore = if let Some(instance_ctx) = self.instance_ctx {
init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
.await
.expect("init instance local disks");
ECStore::new_with_instance_ctx(server_addr, endpoint_pools.clone(), CancellationToken::new(), instance_ctx)
.await
.expect("build instance test ECStore")
} else {
init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
ECStore::new(server_addr, endpoint_pools.clone(), CancellationToken::new())
.await
.expect("build test ECStore")
};
// The production bootstrap only persists pool.bin from the elected
// first cluster node. Test stores intentionally have no cluster
@@ -278,3 +295,86 @@ impl TestECStoreEnvBuilder {
}
}
}
#[cfg(test)]
mod context_tests {
use super::{InstanceContext, TestECStoreEnv};
use crate::ecstore_test_compat::fixture::{read_config, save_config};
use std::sync::Arc;
#[tokio::test]
async fn explicit_context_keeps_stores_isolated() {
let root_a = tempfile::tempdir().expect("create context A disk root");
let root_b = tempfile::tempdir().expect("create context B disk root");
let ctx_a = Arc::new(InstanceContext::new());
let ctx_b = Arc::new(InstanceContext::new());
let env_a = TestECStoreEnv::builder()
.base_dir(root_a.path())
.init_bucket_metadata(false)
.instance_ctx(ctx_a.clone())
.build()
.await;
let env_b = TestECStoreEnv::builder()
.base_dir(root_b.path())
.init_bucket_metadata(false)
.instance_ctx(ctx_b.clone())
.build()
.await;
assert!(env_a.ecstore.instance_endpoints().is_none());
assert!(env_b.ecstore.instance_endpoints().is_none());
ctx_a.set_endpoints(env_a.endpoint_pools.clone());
let endpoints_a = env_a
.ecstore
.instance_endpoints()
.expect("instance A sees its caller-owned context topology");
assert_eq!(
endpoints_a.0[0]
.endpoints
.into_ref()
.first()
.expect("context A first endpoint")
.to_string(),
env_a.disk_paths[0].to_string_lossy(),
);
assert!(
env_b.ecstore.instance_endpoints().is_none(),
"publishing context A must not publish context B"
);
ctx_b.set_endpoints(env_b.endpoint_pools.clone());
let endpoints_b = env_b
.ecstore
.instance_endpoints()
.expect("instance B sees its caller-owned context topology");
assert_eq!(
endpoints_b.0[0]
.endpoints
.into_ref()
.first()
.expect("context B first endpoint")
.to_string(),
env_b.disk_paths[0].to_string_lossy(),
);
assert_eq!(
env_a.ecstore.instance_endpoints().expect("context A remains published").0[0]
.endpoints
.into_ref()
.first()
.expect("context A retained endpoint")
.to_string(),
env_a.disk_paths[0].to_string_lossy(),
);
let path = "config/test-context-isolation.json";
let bytes_a = br#"{"context":"a"}"#.to_vec();
let bytes_b = br#"{"context":"b"}"#.to_vec();
save_config(env_a.ecstore.clone(), path, bytes_a.clone())
.await
.expect("write context A config");
save_config(env_b.ecstore.clone(), path, bytes_b.clone())
.await
.expect("write context B config");
assert_eq!(read_config(env_a.ecstore.clone(), path).await.expect("read context A config"), bytes_a);
assert_eq!(read_config(env_b.ecstore.clone(), path).await.expect("read context B config"), bytes_b);
}
}
@@ -55,11 +55,9 @@ pub const MINIO_INTERNAL_ENCRYPTION_KMS_CONTEXT_HEADER: &str = "X-Minio-Internal
/// Reserved RustFS-branded twin of the MinIO-internal SSE key family.
///
/// No RustFS writer emits these keys today — the SSE writer persists the
/// MinIO-branded `X-Minio-Internal-Server-Side-Encryption-*` keys verbatim for
/// interoperability — but redaction (`rustfs_filemeta`) and replication
/// stripping treat the family as sensitive so that a future or third-party
/// writer cannot leak sealed material through the reserved names.
/// The SSE writer persists sealed material under the MinIO-branded names for
/// interoperability. Redaction and replication stripping also cover this twin
/// family, including the encrypted ETag format marker.
pub const RUSTFS_INTERNAL_ENCRYPTION_PREFIX: &str = "x-rustfs-internal-server-side-encryption-";
pub const REPLICATION_SSEC_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Ssec-Algorithm";
@@ -71,6 +69,15 @@ pub const REPLICATION_SSE_SEAL_ALGORITHM_HEADER: &str = "X-Rustfs-Replication-Se
pub const REPLICATION_SSE_SEALED_KEY_HEADER: &str = "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key";
pub const REPLICATION_ENCRYPTED_MULTIPART_HEADER: &str = "X-Rustfs-Replication-Encrypted-Multipart";
pub const SUFFIX_OPAQUE_ENCRYPTED_ETAG: &str = "server-side-encryption-opaque-etag";
/// Old encrypted objects can carry a plaintext MD5. Keyless listings expose
/// an encrypted ETag only when its producer recorded the protected format.
pub fn etag_can_be_listed(metadata: &HashMap<String, String>) -> bool {
!metadata.keys().any(|key| is_replication_stripped_encryption_key(key))
|| super::metadata_compat::get_consistent_str(metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG) == Some("v1")
}
/// Stored SSE-C metadata keys and the wire names they replicate under.
///
/// Source keys must match what `encryption_material_to_metadata` persists; the
@@ -307,6 +314,28 @@ pub fn stored_managed_encryption_key(metadata: &HashMap<String, String>) -> Opti
mod tests {
use super::*;
#[test]
fn keyless_listings_hide_unproven_encrypted_etags() {
assert!(etag_can_be_listed(&HashMap::from([("content-type".to_owned(), "text/plain".to_owned())])));
for key in [
SSEC_ALGORITHM_HEADER,
INTERNAL_ENCRYPTION_KEY_HEADER,
MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER,
MINIO_INTERNAL_ENCRYPTION_KMS_DATA_KEY_HEADER,
"X-AMZ-SERVER-SIDE-ENCRYPTION",
"x-rustfs-internal-server-side-encryption-sealed-key",
] {
let mut metadata = HashMap::from([(key.to_owned(), "encryption-metadata".to_owned())]);
assert!(!etag_can_be_listed(&metadata), "legacy encrypted metadata: {key}");
super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "unknown".to_owned());
assert!(!etag_can_be_listed(&metadata), "unknown producer format: {key}");
super::super::metadata_compat::insert_str(&mut metadata, SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
assert!(etag_can_be_listed(&metadata), "protected encrypted metadata: {key}");
metadata.insert("x-minio-internal-server-side-encryption-opaque-etag".to_owned(), "unknown".to_owned());
assert!(!etag_can_be_listed(&metadata), "conflicting format markers: {key}");
}
}
#[test]
fn ssec_transport_projection_retains_only_redacted_reader_headers() {
let mut headers = http::HeaderMap::new();
+4
View File
@@ -49,6 +49,10 @@ rustfs-rio.workspace = true
tokio = { workspace = true, features = ["io-util", "macros", "rt"] }
thiserror = { workspace = true }
[target.'cfg(target_os = "macos")'.dependencies]
# Keep XZ decoding self-contained in macOS binaries instead of linking Homebrew liblzma.
liblzma = { version = "0.4.8", features = ["static"] }
[dev-dependencies]
rustfs-tokio-tar = { workspace = true }
futures = { workspace = true }
+3
View File
@@ -21,6 +21,9 @@ For bounded local read-backend startup and cancellation, see
For legacy protection assessment and bounded protected copies, see [Object integrity inventory, audit, and migration](operations/shard-integrity-audit.md).
For encrypted listing compatibility and older stored fingerprints, see
[Encrypted object ETags](operations/encrypted-etags.md).
Operational runbooks live under [`operations/`](operations/). Replication
operators should start with:
+44
View File
@@ -0,0 +1,44 @@
# Encrypted object ETags
**Use this when:** upgrading encrypted storage or investigating missing ETags
in object listings after an upgrade.
New encrypted writes use a keyed fingerprint for the ETag rather than the
plaintext MD5. Single-part and part ETags are 32-character hexadecimal values;
completed multipart ETags retain their digest-and-part-count form. They stay
stable for the stored object. Request MD5 and checksum validation still apply
to the original plaintext. Encryption and decryption formats do not change.
## Listings and older objects
ListObjects, ListObjectsV2, and ListObjectVersions omit the optional ETag field
for encrypted objects whose metadata does not attest to the protected format.
This includes encrypted objects written before the upgrade, preserved source
ETags without that proof, and completed encrypted multipart uploads. New
single-part encrypted writes carry the internal format marker and expose their
protected ETag in listings. Unencrypted listings retain their existing ETags.
Existing persistent listing snapshots are invalidated; the provider falls
back to verified storage until a new snapshot is built.
Clients that require an ETag for these older or multipart objects must issue an
authorized HeadObject request, including the SSE-C key when required. GET,
HEAD, and conditional requests continue to use the object's stored ETag.
## Existing data and migration
The upgrade does not rewrite existing object metadata or erase fingerprints
already copied to backups, logs, caches, or replicas. Restrict access to those
copies separately. Rolling back to an older server also restores its listing
behavior, so keep the upgraded listing protection on every serving node.
To obtain a protected ETag, read the object with its encryption credentials and
write it through the upgraded encryption path. Verify the resulting plaintext
and application references before retiring an old copy. A server-side copy
that preserves the source ETag is insufficient.
In a versioned bucket, a new write creates a new version; historical versions
keep their original metadata. Object Lock retention and legal holds still
apply. Do not bypass them or delete retained versions to migrate an ETag.
There is no automatic in-place migration of retained metadata in this upgrade.
Unfinished uploads also keep their original part metadata. Restart an upload
through the upgraded path when its legacy part fingerprints need to be replaced.
+1
View File
@@ -389,6 +389,7 @@ symbolic-demangle = { version = "=13.9.0", default-features = false, features =
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
rustfs-iam = { workspace = true, features = ["test-util"] }
serial_test = { workspace = true }
tempfile = { workspace = true }
aws-config = { workspace = true }
+13 -13
View File
@@ -17,12 +17,12 @@
use crate::admin::auth::authorize_admin_request;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::current_object_store_handle;
use crate::admin::storage_api::s3::{self, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use crate::server::ADMIN_PREFIX;
use crate::site_replication::{site_replication_enabled, with_site_replication_bucket_mutation_lock};
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_policy::policy::action::{Action, AdminAction};
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
use tracing::info;
use uuid::Uuid;
@@ -63,23 +63,23 @@ impl Operation for RecoverOrphanedBucketHandler {
.credentials
.as_ref()
.map(|credentials| credentials.secret_key.clone())
.ok_or_else(|| s3_error!(InvalidRequest, "authentication required"))?;
.ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "authentication required"))?;
let bucket = params
.get("bucket")
.filter(|bucket| !bucket.is_empty())
.ok_or_else(|| s3_error!(InvalidRequest, "bucket name is required"))?;
.ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "bucket name is required"))?;
let body = req
.input
.store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE)
.await
.map_err(|_| s3_error!(InvalidRequest, "failed to read recovery request body"))?;
.map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "failed to read recovery request body"))?;
let request: RecoverOrphanedBucketRequest = serde_json::from_slice(&body)
.map_err(|_| s3_error!(InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
.map_err(|_| s3::error(S3ErrorCode::InvalidRequest, "expected JSON with a non-nil expectedIncarnationId"))?;
if request.expected_incarnation_id.is_nil() {
return Err(s3_error!(InvalidRequest, "expectedIncarnationId must be non-nil"));
return Err(s3::error(S3ErrorCode::InvalidRequest, "expectedIncarnationId must be non-nil"));
}
let Some(store) = current_object_store_handle() else {
return Err(s3_error!(InternalError, "object store is not initialized"));
return Err(s3::error(S3ErrorCode::InternalError, "object store is not initialized"));
};
let operation_bucket = bucket.to_owned();
@@ -91,21 +91,21 @@ impl Operation for RecoverOrphanedBucketHandler {
let lock_bucket = operation_bucket.clone();
with_site_replication_bucket_mutation_lock(lock_store, &lock_bucket, move || async move {
if site_replication_enabled().await? {
return Err(s3_error!(
OperationAborted,
"orphaned bucket recovery is local; reconcile every site before recreating this bucket"
return Err(s3::error(
S3ErrorCode::OperationAborted,
"orphaned bucket recovery is local; reconcile every site before recreating this bucket",
));
}
operation_store
.recover_orphaned_bucket(&operation_bucket, expected_incarnation)
.await
.map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))
.map_err(|error| S3Error::from(crate::error::ApiError::from(error)))
})
.await??;
Ok::<(), s3s::S3Error>(())
Ok::<(), S3Error>(())
})
.await
.map_err(|error| s3s::S3Error::from(crate::error::ApiError::from(error)))?;
.map_err(|error| S3Error::from(crate::error::ApiError::from(error)))?;
completed?;
info!(
+3 -145
View File
@@ -15,7 +15,7 @@
use crate::admin::auth::authorize_admin_request;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{
current_deployment_id, current_federated_identity_runtime, current_iam_handle, current_object_store_handle, current_region,
current_deployment_id, current_iam_handle, current_object_store_handle, current_oidc_config_query, current_region,
current_replication_pool_handle, current_replication_stats_handle, current_server_config, current_token_signing_key,
object_store_from_req,
};
@@ -2086,40 +2086,11 @@ fn open_id_provider_settings(config: &rustfs_iam::oidc::OidcProviderConfig) -> O
}
}
fn open_id_settings_from_service(service: &rustfs_iam::federation::FederatedIdentityService, region: String) -> OpenIDSettings {
let providers = service
.list_providers()
.into_iter()
.filter_map(|provider| {
let config = service.get_provider_config(&provider.provider_id)?;
Some((provider.provider_id, open_id_provider_settings(config)))
})
.collect();
open_id_settings(providers, region)
}
fn open_id_settings_from_runtime(
oidc_config_query: Option<&dyn rustfs_iam::federation::oidc::OidcConfigQuery>,
service: Option<&rustfs_iam::federation::FederatedIdentityService>,
region: String,
) -> OpenIDSettings {
if let Some(oidc_config_query) = oidc_config_query {
return open_id_settings_from_snapshot(&oidc_config_query.config_snapshot(), region);
}
service
.map(|service| open_id_settings_from_service(service, region))
.unwrap_or_default()
}
fn local_idp_settings() -> IDPSettings {
let mut settings = IDPSettings::default();
let region = current_region().map(|region| region.to_string()).unwrap_or_default();
settings.open_id = current_federated_identity_runtime()
.map(|(service, oidc_config_query)| {
open_id_settings_from_runtime(oidc_config_query.as_deref(), Some(service.as_ref()), region)
})
settings.open_id = current_oidc_config_query()
.map(|query| open_id_settings_from_snapshot(&query.config_snapshot(), region))
.unwrap_or_default();
let (ldap, ldap_configs) = load_ldap_idp_settings();
@@ -11444,82 +11415,6 @@ mod tests {
}
}
struct LegacyOidcProvider {
configs: Vec<rustfs_iam::oidc::OidcProviderConfig>,
}
struct SnapshotOidcConfigQuery {
snapshot: rustfs_iam::oidc::OidcConfigSnapshot,
}
impl rustfs_iam::federation::oidc::OidcConfigQuery for SnapshotOidcConfigQuery {
fn config_snapshot(&self) -> rustfs_iam::oidc::OidcConfigSnapshot {
self.snapshot.clone()
}
}
#[async_trait::async_trait]
impl rustfs_iam::federation::FederatedIdentityProvider for LegacyOidcProvider {
fn has_providers(&self) -> bool {
!self.configs.is_empty()
}
fn list_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
self.configs
.iter()
.map(|config| rustfs_iam::federation::FederatedProviderView {
provider_id: config.id.clone(),
display_name: config.display_name.clone(),
})
.collect()
}
fn list_visible_providers(&self) -> Vec<rustfs_iam::federation::FederatedProviderView> {
self.list_providers()
}
fn provider_config(&self, provider_id: &str) -> Option<&rustfs_iam::oidc::OidcProviderConfig> {
self.configs.iter().find(|config| config.id == provider_id)
}
async fn authorize_url(
&self,
_provider_id: &str,
_redirect_uri: &str,
_redirect_after: Option<String>,
) -> rustfs_iam::federation::Result<String> {
unreachable!("site replication compatibility test does not authorize")
}
async fn exchange_code(
&self,
_state: &str,
_code: &str,
_redirect_uri: &str,
) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedCodeExchange> {
unreachable!("site replication compatibility test does not exchange codes")
}
async fn verify_web_identity_token(
&self,
_jwt: &str,
) -> rustfs_iam::federation::Result<rustfs_iam::federation::FederatedAuthorization> {
unreachable!("site replication compatibility test does not verify tokens")
}
async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> rustfs_iam::federation::Result<String> {
unreachable!("site replication compatibility test does not create logout tokens")
}
async fn build_logout_url(
&self,
_logout_token: &str,
_post_logout_redirect_uri: &str,
) -> rustfs_iam::federation::Result<Option<String>> {
unreachable!("site replication compatibility test does not build logout URLs")
}
}
fn ldap_settings_kvs() -> rustfs_config::server_config::KVS {
rustfs_config::server_config::KVS(vec![
rustfs_config::server_config::KV {
@@ -11692,43 +11587,6 @@ mod tests {
);
}
#[test]
fn service_only_oidc_runtime_preserves_site_replication_settings() {
let default = oidc_snapshot_provider("default", "client-a");
let corp = oidc_snapshot_provider("corp", "client-b");
let expected = open_id_settings_from_snapshot(
&rustfs_iam::oidc::OidcConfigSnapshot::new(vec![corp.clone(), default.clone()]),
"eu-site-1".to_string(),
);
let service = rustfs_iam::federation::FederatedIdentityService::new(
rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
configs: vec![corp.config, default.config],
})),
);
assert_eq!(
serde_json::to_value(open_id_settings_from_runtime(None, Some(&service), "eu-site-1".to_string(),))
.expect("serialize legacy OpenID settings"),
serde_json::to_value(expected).expect("serialize snapshot OpenID settings"),
);
}
#[test]
fn oidc_query_takes_precedence_over_service_only_compatibility() {
let query = SnapshotOidcConfigQuery {
snapshot: rustfs_iam::oidc::OidcConfigSnapshot::new(vec![oidc_snapshot_provider("default", "query-client")]),
};
let service = rustfs_iam::federation::FederatedIdentityService::new(
rustfs_iam::federation::FederatedIdentityRegistry::new(Arc::new(LegacyOidcProvider {
configs: vec![oidc_snapshot_provider("default", "service-client").config],
})),
);
let settings = open_id_settings_from_runtime(Some(&query), Some(&service), "eu-site-1".to_string());
assert_eq!(settings.claim_provider.client_id, "query-client");
}
// A whole provider object present on one side only must not spill its
// credential-derived leaves into the rejection message.
#[test]
+5 -6
View File
@@ -23,12 +23,11 @@ use crate::app::object_usecase::DefaultObjectUsecase;
use crate::runtime_sources as root_runtime_sources;
pub(crate) use crate::runtime_sources::{
AppContext, ServerContextSlot, current_action_credentials, current_boot_time, current_bucket_metadata_handle,
current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_runtime,
current_federated_identity_service, current_iam_handle, current_kms_runtime_service_manager,
current_notification_system_for_context, current_object_data_cache_handle_for_context,
current_object_store_handle_for_context, current_ready_iam_handle, current_region, current_replication_pool_handle,
current_replication_stats_handle, current_replication_stats_handle_for_context, current_server_config_for_context,
current_token_signing_key,
current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_service,
current_iam_handle, current_kms_runtime_service_manager, current_notification_system_for_context,
current_object_data_cache_handle_for_context, current_object_store_handle_for_context, current_oidc_config_query,
current_ready_iam_handle, current_region, current_replication_pool_handle, current_replication_stats_handle,
current_replication_stats_handle_for_context, current_server_config_for_context, current_token_signing_key,
};
#[cfg(test)]
pub(crate) use crate::runtime_sources::{
+93 -2
View File
@@ -1086,7 +1086,11 @@ fn build_list_object_versions_metadata_output(
size: Some(object.size),
version_id: Some(version_id),
is_latest: Some(object.is_latest),
e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
e_tag: object
.etag
.as_ref()
.filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
.map(|etag| to_s3s_etag(etag)),
storage_class: Some(ObjectVersionStorageClass::from(
object
.storage_class
@@ -1173,7 +1177,11 @@ fn build_list_objects_v2_metadata_output(
key: Some(encode_list_objects_v2_value(&object.name, encoding_type)),
last_modified: object.mod_time.map(Timestamp::from),
size: Some(object.get_actual_size_or_physical()),
e_tag: object.etag.clone().map(|etag| to_s3s_etag(&etag)),
e_tag: object
.etag
.as_ref()
.filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined))
.map(|etag| to_s3s_etag(etag)),
storage_class: Some(ObjectStorageClass::from(
object
.storage_class
@@ -4548,6 +4556,89 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::InternalError);
}
#[test]
fn keyless_object_and_version_listings_do_not_expose_legacy_encrypted_fingerprints() {
let legacy_md5 = "fbfcbcc6d2f035411e6268c8b119593e";
let mut protected_metadata =
HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned())]);
rustfs_utils::http::metadata_compat::insert_str(
&mut protected_metadata,
rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
"v1".to_owned(),
);
let objects = vec![
ObjectInfo {
name: "legacy-encrypted".to_owned(),
etag: Some(legacy_md5.to_owned()),
user_defined: Arc::new(HashMap::from([(
"x-amz-server-side-encryption-customer-algorithm".to_owned(),
"AES256".to_owned(),
)])),
..Default::default()
},
ObjectInfo {
name: "protected-encrypted".to_owned(),
etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
user_defined: Arc::new(protected_metadata),
..Default::default()
},
ObjectInfo {
name: "plain".to_owned(),
etag: Some("9876543210abcdef9876543210abcdef".to_owned()),
..Default::default()
},
];
let listings = build_list_objects_v2_metadata_output(
ListObjectsV2Info {
objects: objects.clone(),
..Default::default()
},
"demo-bucket",
&ListObjectsV2Params {
prefix: String::new(),
max_keys: 1000,
delimiter: None,
response_start_after: None,
start_after_for_query: None,
response_continuation_token: None,
decoded_continuation_token: None,
},
None,
false,
&HashMap::new(),
);
assert!(listings.contents[0].object.e_tag.is_none());
assert!(listings.contents[1].object.e_tag.is_some());
assert!(listings.contents[2].object.e_tag.is_some());
let xml = String::from_utf8(serialize_config(&listings).expect("serialize objects")).expect("UTF-8 objects");
assert!(!xml.contains(legacy_md5));
let versions = build_list_object_versions_metadata_output(
ListObjectVersionsInfo {
objects,
..Default::default()
},
"demo-bucket",
&ListObjectVersionsParams {
prefix: String::new(),
max_keys: 1000,
delimiter: None,
key_marker: None,
version_id_marker: None,
},
None,
&HashMap::new(),
);
for (index, entry) in versions.entries.iter().enumerate() {
let ListObjectVersionMetadataEntry::Version(version, _) = entry else {
panic!("expected object version");
};
assert_eq!(version.e_tag.is_some(), index != 0);
}
let xml = String::from_utf8(serialize_config(&versions).expect("serialize versions")).expect("UTF-8 versions");
assert!(!xml.contains(legacy_md5));
}
#[test]
fn build_list_object_versions_metadata_output_maps_metadata_and_preserves_entry_order() {
use time::macros::datetime;
+6 -32
View File
@@ -103,11 +103,6 @@ pub fn resolve_oidc_config_query() -> Option<Arc<dyn OidcConfigQuery>> {
resolve_oidc_config_query_with(get_global_app_context())
}
/// Resolve the paired federation service and OIDC configuration query from one runtime observation.
pub(crate) fn resolve_federated_identity_runtime() -> Option<FederatedIdentityRuntimeSnapshot> {
resolve_federated_identity_runtime_with(get_global_app_context())
}
/// Publish the initialized federation service and OIDC configuration query.
pub fn publish_federated_identity_runtime(
service: Arc<FederatedIdentityService>,
@@ -116,11 +111,6 @@ pub fn publish_federated_identity_runtime(
handles::publish_default_federated_identity_runtime(service, oidc_config_query)
}
/// Publish a service-only federation runtime for compatibility integrations.
pub fn publish_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
handles::publish_default_federated_identity_service(service)
}
/// Resolve a ready IAM system handle using AppContext-first precedence.
pub fn resolve_ready_iam_handle() -> rustfs_iam::error::Result<Arc<IamSys<ObjectStore>>> {
if let Some(context) = get_global_app_context() {
@@ -360,6 +350,7 @@ fn resolve_oidc_config_query_with(context: Option<Arc<AppContext>>) -> Option<Ar
context.and_then(|context| context.federated_identity().oidc_config_query())
}
#[cfg(test)]
fn resolve_federated_identity_runtime_with(context: Option<Arc<AppContext>>) -> Option<FederatedIdentityRuntimeSnapshot> {
context.and_then(|context| context.federated_identity().runtime_snapshot())
}
@@ -554,7 +545,7 @@ mod tests {
use async_trait::async_trait;
use rustfs_iam::{
federation::{
FederatedIdentityRegistry, FederatedIdentityService,
CoreFederatedAuthorizationMapper, FederatedIdentityService,
oidc::{OidcConfigQuery, StandardOidcAdapter},
},
oidc::OidcSys,
@@ -617,25 +608,11 @@ mod tests {
}
impl FederatedIdentityInterface for TestFederatedIdentityInterface {
fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
self.runtime
.read()
.ok()
.and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.service)))
}
fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
self.runtime
.read()
.ok()
.and_then(|runtime| runtime.as_ref().map(|runtime| Arc::clone(&runtime.oidc_config_query)))
}
fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
self.runtime.read().ok().and_then(|runtime| {
runtime
.as_ref()
.map(|runtime| (Arc::clone(&runtime.service), Some(Arc::clone(&runtime.oidc_config_query))))
.map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
})
}
@@ -654,7 +631,8 @@ mod tests {
fn test_federated_identity_runtime(oidc: OidcSys) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
let oidc_config_query = adapter.clone();
let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
let service = Arc::new(adapter.into_service(mapper));
(service, oidc_config_query)
}
@@ -1206,11 +1184,7 @@ mod tests {
let (resolved_oidc, resolved_oidc_query) =
resolve_federated_identity_runtime_with(Some(context.clone())).expect("context federated identity runtime");
assert!(Arc::ptr_eq(&resolved_oidc, &context_oidc));
assert!(
resolved_oidc_query
.as_ref()
.is_some_and(|query| Arc::ptr_eq(query, &context_oidc_query))
);
assert!(Arc::ptr_eq(&resolved_oidc_query, &context_oidc_query));
assert!(publish_federated_identity_runtime_with(
Some(context.clone()),
fallback_oidc.clone(),
-5
View File
@@ -169,11 +169,6 @@ impl AppContext {
self.federated_identity.publish_runtime(service, oidc_config_query)
}
/// Publish a service-only federation runtime for compatibility integrations.
pub fn publish_federated_identity_service(&self, service: Arc<FederatedIdentityService>) -> bool {
self.federated_identity.publish_handle(service)
}
pub fn kms(&self) -> Arc<dyn KmsInterface> {
self.kms.clone()
}
+12 -74
View File
@@ -87,24 +87,15 @@ pub struct FederatedIdentityHandle {
struct FederatedIdentityRuntime {
service: Arc<FederatedIdentityService>,
oidc_config_query: Option<Arc<dyn OidcConfigQuery>>,
oidc_config_query: Arc<dyn OidcConfigQuery>,
}
impl FederatedIdentityHandle {
pub fn new(service: Option<Arc<FederatedIdentityService>>) -> Self {
Self {
runtime: StdRwLock::new(service.map(|service| FederatedIdentityRuntime {
service,
oidc_config_query: None,
})),
}
}
pub fn with_runtime(runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>) -> Self {
Self {
runtime: StdRwLock::new(runtime.map(|(service, oidc_config_query)| FederatedIdentityRuntime {
service,
oidc_config_query: Some(oidc_config_query),
oidc_config_query,
})),
}
}
@@ -112,45 +103,26 @@ impl FederatedIdentityHandle {
impl Default for FederatedIdentityHandle {
fn default() -> Self {
Self::new(None)
Self::with_runtime(None)
}
}
impl FederatedIdentityInterface for FederatedIdentityHandle {
fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
self.runtime_snapshot().map(|(service, _)| service)
}
fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
self.runtime_snapshot().and_then(|(_, oidc_config_query)| oidc_config_query)
}
fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
self.runtime.read().ok().and_then(|runtime| {
runtime
.as_ref()
.map(|runtime| (Arc::clone(&runtime.service), runtime.oidc_config_query.as_ref().map(Arc::clone)))
.map(|runtime| (Arc::clone(&runtime.service), Arc::clone(&runtime.oidc_config_query)))
})
}
fn publish_handle(&self, service: Arc<FederatedIdentityService>) -> bool {
let Ok(mut published_runtime) = self.runtime.write() else {
return false;
};
*published_runtime = Some(FederatedIdentityRuntime {
service,
oidc_config_query: None,
});
true
}
fn publish_runtime(&self, service: Arc<FederatedIdentityService>, oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
let Ok(mut published_runtime) = self.runtime.write() else {
return false;
};
*published_runtime = Some(FederatedIdentityRuntime {
service,
oidc_config_query: Some(oidc_config_query),
oidc_config_query,
});
true
}
@@ -639,15 +611,6 @@ pub fn publish_default_federated_identity_runtime(
default_federated_identity_handle().publish_runtime(service, oidc_config_query)
}
/// Publish a service-only federation runtime for compatibility integrations.
pub fn publish_default_federated_identity_service(service: Arc<FederatedIdentityService>) -> bool {
default_federated_identity_handle().publish_handle(service)
}
pub fn federated_identity_interface(service: Option<Arc<FederatedIdentityService>>) -> Arc<dyn FederatedIdentityInterface> {
Arc::new(FederatedIdentityHandle::new(service))
}
pub fn federated_identity_runtime_interface(
runtime: Option<(Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>)>,
) -> Arc<dyn FederatedIdentityInterface> {
@@ -686,14 +649,13 @@ pub fn default_buffer_config_interface() -> Arc<dyn BufferConfigInterface> {
mod tests {
use super::{
KmsRuntimeHandle, KmsServiceManager, ServerConfigHandle, default_federated_identity_interface,
federated_identity_interface, federated_identity_runtime_interface, publish_default_federated_identity_runtime,
runtime_sources,
federated_identity_runtime_interface, publish_default_federated_identity_runtime, runtime_sources,
};
use crate::app::context::interfaces::{KmsRuntimeInterface, ServerConfigInterface};
use rustfs_config::server_config::Config;
use rustfs_iam::{
federation::{
FederatedIdentityRegistry, FederatedIdentityService,
CoreFederatedAuthorizationMapper, FederatedIdentityService,
oidc::{OidcConfigQuery, StandardOidcAdapter},
},
oidc::OidcSys,
@@ -705,7 +667,8 @@ mod tests {
let oidc = OidcSys::empty().expect("empty OIDC configuration should be valid");
let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
let oidc_config_query = adapter.clone();
let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(adapter)));
let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
let service = Arc::new(adapter.into_service(mapper));
(service, oidc_config_query)
}
@@ -726,7 +689,7 @@ mod tests {
assert!(Arc::ptr_eq(&resolved_query, &first_query));
let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("initial federated identity runtime");
assert!(Arc::ptr_eq(&resolved_service, &first));
assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &first_query)));
assert!(Arc::ptr_eq(&resolved_query, &first_query));
assert!(interface.publish_runtime(second.clone(), second_query.clone()));
assert!(
@@ -738,36 +701,11 @@ mod tests {
assert!(interface.oidc_config_query().is_some());
let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("replacement federated identity runtime");
assert!(Arc::ptr_eq(&resolved_service, &second));
assert!(resolved_query.as_ref().is_some_and(|query| Arc::ptr_eq(query, &second_query)));
}
#[test]
fn federated_identity_handle_preserves_service_only_compatibility() {
let (service, _) = test_federated_identity_runtime();
let (replacement, _) = test_federated_identity_runtime();
let interface = federated_identity_interface(Some(service.clone()));
assert!(
interface
.handle()
.as_ref()
.is_some_and(|resolved| Arc::ptr_eq(resolved, &service))
);
assert!(interface.oidc_config_query().is_none());
assert!(interface.publish_handle(replacement.clone()));
assert!(
interface
.handle()
.as_ref()
.is_some_and(|resolved| Arc::ptr_eq(resolved, &replacement))
);
assert!(interface.oidc_config_query().is_none());
let (resolved_service, resolved_query) = interface.runtime_snapshot().expect("service-only federated identity runtime");
assert!(Arc::ptr_eq(&resolved_service, &replacement));
assert!(resolved_query.is_none());
assert!(Arc::ptr_eq(&resolved_query, &second_query));
}
#[test]
#[serial_test::serial(federated_identity_runtime)]
fn default_federated_identity_handle_shares_early_publish_and_replacement() {
let (first, first_query) = test_federated_identity_runtime();
let (second, second_query) = test_federated_identity_runtime();
+10 -13
View File
@@ -48,25 +48,22 @@ pub trait IamInterface: Send + Sync {
}
/// Published federation runtime handles for authentication and OIDC configuration consumers.
pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Option<Arc<dyn OidcConfigQuery>>);
pub type FederatedIdentityRuntimeSnapshot = (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>);
pub trait FederatedIdentityInterface: Send + Sync {
fn handle(&self) -> Option<Arc<FederatedIdentityService>>;
fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot>;
fn handle(&self) -> Option<Arc<FederatedIdentityService>> {
self.runtime_snapshot().map(|(service, _)| service)
}
fn oidc_config_query(&self) -> Option<Arc<dyn OidcConfigQuery>> {
None
self.runtime_snapshot().map(|(_, oidc_config_query)| oidc_config_query)
}
fn runtime_snapshot(&self) -> Option<FederatedIdentityRuntimeSnapshot> {
self.handle().map(|service| (service, self.oidc_config_query()))
}
/// Compatibility publication hook for service-only implementations.
/// Standard OIDC runtimes publish the paired query through
/// [`Self::publish_runtime`].
fn publish_handle(&self, _service: Arc<FederatedIdentityService>) -> bool {
fn publish_runtime(&self, _service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
false
}
fn publish_runtime(&self, service: Arc<FederatedIdentityService>, _oidc_config_query: Arc<dyn OidcConfigQuery>) -> bool {
self.publish_handle(service)
}
}
/// KMS interface for application-layer use-cases.
+9
View File
@@ -511,6 +511,15 @@ impl DefaultObjectUsecase {
// cannot flood the log.
let mut bulk_denial_logged = false;
for (idx, obj_id) in delete.objects.iter().enumerate() {
if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
delete_results[idx].error = Some(s3s::dto::Error {
code: Some(err.code().as_str().to_string()),
key: Some(obj_id.key.clone()),
message: Some(err.to_string()),
version_id: obj_id.version_id.clone(),
});
continue;
}
let raw_version_id = obj_id.version_id.clone();
let (version_id, version_uuid) = match normalize_delete_objects_version_id(raw_version_id.clone()) {
Ok(parsed) => parsed,
+30 -79
View File
@@ -35,7 +35,6 @@ pub(crate) use context::{
resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_buffer_config as current_buffer_config,
resolve_daily_tier_stats as current_daily_tier_stats, resolve_deployment_id as current_deployment_id,
resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle,
resolve_federated_identity_runtime as current_federated_identity_runtime,
resolve_federated_identity_service as current_federated_identity_service, resolve_iam_handle as current_iam_handle,
resolve_iam_ready as current_iam_ready, resolve_internode_metrics as current_internode_metrics,
resolve_kms_runtime_service_manager as current_kms_runtime_service_manager,
@@ -47,6 +46,7 @@ pub(crate) use context::{
resolve_object_data_cache_handle_for_context as current_object_data_cache_handle_for_context,
resolve_object_store_handle as current_object_store_handle,
resolve_object_store_handle_for_context as current_object_store_handle_for_context,
resolve_oidc_config_query as current_oidc_config_query,
resolve_or_init_kms_runtime_service_manager as current_or_init_kms_runtime_service_manager,
resolve_outbound_tls_generation as current_outbound_tls_generation, resolve_outbound_tls_state as current_outbound_tls_state,
resolve_performance_metrics as current_performance_metrics, resolve_ready_iam_handle as current_ready_iam_handle,
@@ -63,13 +63,13 @@ pub(crate) fn publish_federated_identity_runtime(
service: Arc<FederatedIdentityService>,
oidc_config_query: Arc<dyn OidcConfigQuery>,
) -> bool {
warn_on_browser_redirect_fallback(&service);
warn_on_browser_redirect_fallback(service.has_providers());
context::publish_federated_identity_runtime(service, oidc_config_query)
}
fn warn_on_browser_redirect_fallback(service: &FederatedIdentityService) {
fn warn_on_browser_redirect_fallback(has_providers: bool) {
let browser_redirect_url = rustfs_utils::get_env_opt_str(ENV_RUSTFS_BROWSER_REDIRECT_URL);
if service.has_providers() && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
if has_providers && browser_redirect_url.as_deref().is_none_or(|value| value.trim().is_empty()) {
warn!(
event = EVENT_OIDC_BROWSER_REDIRECT_FALLBACK,
component = LOG_COMPONENT_MAIN,
@@ -103,76 +103,19 @@ pub(crate) fn current_app_context() -> Option<Arc<AppContext>> {
#[cfg(test)]
mod tests {
use super::*;
use rustfs_iam::federation::oidc::OidcConfigQuery;
use rustfs_iam::federation::{
FederatedAuthorization, FederatedCodeExchange, FederatedIdentityProvider, FederatedIdentityRegistry,
FederatedProviderView, FederatedRedirectPolicy, Result as FederationResult,
use rustfs_iam::{
federation::{
CoreFederatedAuthorizationMapper,
oidc::{OidcConfigQuery, StandardOidcAdapter},
},
oidc::{OidcSys, make_test_sys, test_config},
};
use rustfs_iam::oidc::OidcConfigSnapshot;
use std::{
io::{self, Write},
sync::Mutex,
};
use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
struct TestProvider {
has_providers: bool,
}
#[async_trait::async_trait]
impl FederatedIdentityProvider for TestProvider {
fn has_providers(&self) -> bool {
self.has_providers
}
fn list_providers(&self) -> Vec<FederatedProviderView> {
Vec::new()
}
fn list_visible_providers(&self) -> Vec<FederatedProviderView> {
Vec::new()
}
fn redirect_policy(&self, _provider_id: &str) -> Option<FederatedRedirectPolicy> {
None
}
async fn authorize_url(
&self,
_provider_id: &str,
_redirect_uri: &str,
_redirect_after: Option<String>,
) -> FederationResult<String> {
unreachable!("startup publication test does not authorize")
}
async fn exchange_code(&self, _state: &str, _code: &str, _redirect_uri: &str) -> FederationResult<FederatedCodeExchange> {
unreachable!("startup publication test does not exchange codes")
}
async fn verify_web_identity_token(&self, _jwt: &str) -> FederationResult<FederatedAuthorization> {
unreachable!("startup publication test does not verify tokens")
}
async fn create_logout_token(&self, _provider_id: &str, _id_token: &str) -> FederationResult<String> {
unreachable!("startup publication test does not create logout tokens")
}
async fn build_logout_url(
&self,
_logout_token: &str,
_post_logout_redirect_uri: &str,
) -> FederationResult<Option<String>> {
unreachable!("startup publication test does not build logout URLs")
}
}
impl OidcConfigQuery for TestProvider {
fn config_snapshot(&self) -> OidcConfigSnapshot {
OidcConfigSnapshot::new(Vec::new())
}
}
#[derive(Clone, Default)]
struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
@@ -197,14 +140,14 @@ mod tests {
}
}
fn test_runtime(has_providers: bool) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
let provider = Arc::new(TestProvider { has_providers });
let service = Arc::new(FederatedIdentityService::new(FederatedIdentityRegistry::new(provider.clone())));
(service, provider)
}
fn capture_startup_publication(has_providers: bool, browser_redirect_url: Option<&str>) -> String {
fn capture_startup_publication(oidc: OidcSys, browser_redirect_url: Option<&str>) -> String {
temp_env::with_var(ENV_RUSTFS_BROWSER_REDIRECT_URL, browser_redirect_url, || {
let adapter = Arc::new(StandardOidcAdapter::new(Arc::new(oidc)));
let oidc_config_query: Arc<dyn OidcConfigQuery> = adapter.clone();
let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
let service = Arc::new(adapter.into_service(mapper));
let expected_service = service.clone();
let expected_oidc_config_query = oidc_config_query.clone();
let logs = CapturedLogs::default();
let captured = logs.0.clone();
let subscriber = Registry::default().with(
@@ -217,8 +160,12 @@ mod tests {
);
tracing::subscriber::with_default(subscriber, || {
let (service, oidc_config_query) = test_runtime(has_providers);
assert!(publish_federated_identity_runtime(service, oidc_config_query));
let (published_service, published_oidc_config_query) = context::default_federated_identity_interface()
.runtime_snapshot()
.expect("published federated identity runtime should resolve");
assert!(Arc::ptr_eq(&published_service, &expected_service));
assert!(Arc::ptr_eq(&published_oidc_config_query, &expected_oidc_config_query));
});
String::from_utf8(captured.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8")
@@ -226,16 +173,20 @@ mod tests {
}
#[test]
#[serial_test::serial(federated_identity_runtime)]
fn oidc_startup_publication_warns_only_for_request_header_fallback() {
let output = capture_startup_publication(true, None);
let output = capture_startup_publication(make_test_sys(vec![test_config("default")]), None);
assert!(output.contains("WARN"), "{output}");
assert!(output.contains(EVENT_OIDC_BROWSER_REDIRECT_FALLBACK), "{output}");
assert!(output.contains(ENV_RUSTFS_BROWSER_REDIRECT_URL), "{output}");
assert!(output.contains("request_host_and_forwarded_proto"), "{output}");
assert!(output.contains("trusted_console_origin_not_configured"), "{output}");
assert!(capture_startup_publication(false, None).is_empty());
assert!(capture_startup_publication(true, Some("https://console.example.com")).is_empty());
assert!(!capture_startup_publication(true, Some(" ")).is_empty());
assert!(capture_startup_publication(OidcSys::empty().expect("empty OIDC system should initialize"), None).is_empty());
assert!(
capture_startup_publication(make_test_sys(vec![test_config("default")]), Some("https://console.example.com"),)
.is_empty()
);
assert!(!capture_startup_publication(make_test_sys(vec![test_config("default")]), Some(" ")).is_empty());
}
}
+165 -18
View File
@@ -140,6 +140,9 @@ struct InternalObjectAuthorization;
#[derive(Clone, Debug)]
struct RequestedObjectLockRetainUntil(OffsetDateTime);
#[derive(Clone, Debug)]
struct RequestedObjectLockRetention(ObjectLockRetention);
#[derive(Clone, Debug)]
struct StagedMultipartPartAuthorization;
@@ -885,6 +888,21 @@ fn authorization_conditions<T>(
}
merge_list_bucket_query_conditions(action, req.uri.query(), &mut conditions);
merge_request_object_tag_conditions(action, &req.headers, &mut conditions)?;
if let Some(RequestedObjectLockRetention(retention)) = req.extensions.get::<RequestedObjectLockRetention>() {
// Standalone retention writes apply the XML body, so headers must not
// supply a different value to the policy decision.
conditions.remove("object-lock-mode");
conditions.remove("object-lock-retain-until-date");
if let Some(mode) = &retention.mode {
conditions.insert("object-lock-mode".to_string(), vec![mode.as_str().to_string()]);
}
if let Some(date) = &retention.retain_until_date {
let date = OffsetDateTime::from(date.clone())
.format(&time::format_description::well_known::Rfc3339)
.map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid retain-until date"))?;
conditions.insert("object-lock-retain-until-date".to_string(), vec![date]);
}
}
merge_object_lock_remaining_retention_days_condition(
req.extensions.get::<RequestedObjectLockRetainUntil>(),
OffsetDateTime::now_utc(),
@@ -908,6 +926,12 @@ pub(crate) fn set_requested_object_lock_retain_until<T>(req: &mut S3Request<T>,
}
}
fn set_requested_object_lock_retention(req: &mut S3Request<PutObjectRetentionInput>) {
let retention = req.input.retention.clone().unwrap_or_default();
set_requested_object_lock_retain_until(req, retention.retain_until_date.as_ref());
req.extensions.insert(RequestedObjectLockRetention(retention));
}
fn merge_object_lock_remaining_retention_days_condition(
requested: Option<&RequestedObjectLockRetainUntil>,
now: OffsetDateTime,
@@ -1123,11 +1147,8 @@ pub(crate) fn delete_object_authorize_action(version_id: Option<&str>) -> Action
/// conflated at the authorization boundary.
///
/// Note: `ActionSet::is_match` still maps a `s3:GetObjectVersion` grant onto a
/// `s3:GetObject` request. That mapping is intentionally left in place until the
/// remaining version-aware read paths (HeadObject, GetObjectAcl, tagging) get the
/// same treatment — see the GHSA-3ppv follow-up audit. It does not re-open this
/// disclosure: it only broadens a Version grant toward current reads, never the
/// reverse.
/// `s3:GetObject` request. It only broadens a Version grant toward current
/// reads, never the reverse.
fn versioned_read_action(version_id: Option<&str>) -> Action {
if version_id.is_some() {
Action::S3Action(S3Action::GetObjectVersionAction)
@@ -1136,6 +1157,20 @@ fn versioned_read_action(version_id: Option<&str>) -> Action {
}
}
fn versioned_tagging_action(action: S3Action, version_id: Option<&str>) -> Action {
let action = if version_id.is_some() {
match action {
S3Action::GetObjectTaggingAction => S3Action::GetObjectVersionTaggingAction,
S3Action::PutObjectTaggingAction => S3Action::PutObjectVersionTaggingAction,
S3Action::DeleteObjectTaggingAction => S3Action::DeleteObjectVersionTaggingAction,
action => action,
}
} else {
action
};
Action::S3Action(action)
}
async fn get_or_fetch_object_tag_conditions<T>(
req: &mut S3Request<T>,
store: &ECStore,
@@ -2560,7 +2595,11 @@ impl S3Access for FS {
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::DeleteObjectTaggingAction)).await
authorize_request(
req,
versioned_tagging_action(S3Action::DeleteObjectTaggingAction, req.input.version_id.as_deref()),
)
.await
}
/// Checks whether the DeleteObjects request has accesses to the resources.
@@ -2881,7 +2920,11 @@ impl S3Access for FS {
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::GetObjectTaggingAction)).await
authorize_request(
req,
versioned_tagging_action(S3Action::GetObjectTaggingAction, req.input.version_id.as_deref()),
)
.await
}
/// Checks whether the GetObjectTorrent request has accesses to the resources.
@@ -2933,7 +2976,7 @@ impl S3Access for FS {
return Ok(());
}
authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?;
req.extensions.insert(source_generation);
Ok(())
}
@@ -3336,12 +3379,7 @@ impl S3Access for FS {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let requested_retain_until = req
.input
.retention
.as_ref()
.and_then(|retention| retention.retain_until_date.clone());
set_requested_object_lock_retain_until(req, requested_retain_until.as_ref());
set_requested_object_lock_retention(req);
authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
// S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
@@ -3362,7 +3400,11 @@ impl S3Access for FS {
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::PutObjectTaggingAction)).await
authorize_request(
req,
versioned_tagging_action(S3Action::PutObjectTaggingAction, req.input.version_id.as_deref()),
)
.await
}
/// Checks whether the PutPublicAccessBlock request has accesses to the resources.
@@ -3481,9 +3523,9 @@ mod tests {
owner_can_bypass_policy_deny, post_object_authorize_action, put_bucket_policy_authorize_action, request_context_from_req,
request_object_store, require_owned_reserved_table_object, retention_write_requested, table_data_plane_admin_action,
table_data_plane_content_mutation, table_data_plane_resource_for_request, table_publication_guard_error,
validate_post_object_success_controls, versioned_read_action,
validate_post_object_success_controls, versioned_read_action, versioned_tagging_action,
};
use super::{remaining_retention_days, set_requested_object_lock_retain_until};
use super::{remaining_retention_days, set_requested_object_lock_retain_until, set_requested_object_lock_retention};
use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
use crate::error::ApiError;
use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
@@ -4285,11 +4327,116 @@ mod tests {
);
}
fn remaining_retention_days_conditions(req: &S3Request<()>, action: Action) -> HashMap<String, Vec<String>> {
#[tokio::test]
async fn versioned_tagging_requires_version_permissions() {
use rustfs_policy::policy::{Args, Policy};
let policy = Policy::parse_config(br#"{
"Version":"2012-10-17",
"Statement":[{"Effect":"Allow","Action":["s3:GetObjectTagging","s3:PutObjectTagging","s3:DeleteObjectTagging"],"Resource":"arn:aws:s3:::bucket/*"}]
}"#).expect("current-object tagging policy should parse");
let conditions = HashMap::new();
let claims = HashMap::new();
let groups = None;
for (current, versioned) in [
(S3Action::GetObjectTaggingAction, S3Action::GetObjectVersionTaggingAction),
(S3Action::PutObjectTaggingAction, S3Action::PutObjectVersionTaggingAction),
(S3Action::DeleteObjectTaggingAction, S3Action::DeleteObjectVersionTaggingAction),
] {
for version in [None, Some("null"), Some("0194e0f1-0000-7000-8000-000000000000")] {
let action = versioned_tagging_action(current.clone(), version);
assert_eq!(
action,
Action::S3Action(if version.is_some() {
versioned.clone()
} else {
current.clone()
})
);
let args = Args {
account: "writer",
groups: &groups,
action,
bucket: "bucket",
conditions: &conditions,
is_owner: false,
object: "object",
claims: &claims,
deny_only: false,
};
assert_eq!(policy.is_allowed(&args).await, version.is_none(), "version selector {version:?}");
}
}
}
fn remaining_retention_days_conditions<T>(req: &S3Request<T>, action: Action) -> HashMap<String, Vec<String>> {
let credentials = rustfs_credentials::Credentials::default();
authorization_conditions(req, &credentials, None, None, None, None, action).expect("conditions should build")
}
#[tokio::test]
async fn retention_policy_checks_body_instead_of_conflicting_headers() {
use rustfs_policy::policy::{Args, Policy};
let mut req = build_request(
PutObjectRetentionInput {
retention: Some(ObjectLockRetention {
mode: Some(ObjectLockRetentionMode::from_static("COMPLIANCE")),
retain_until_date: Some(
Timestamp::parse(TimestampFormat::DateTime, "2035-01-02T00:00:00+23:59").expect("body date should parse"),
),
}),
..Default::default()
},
Method::PUT,
);
req.headers
.insert("x-amz-object-lock-mode", HeaderValue::from_static("GOVERNANCE"));
req.headers
.insert("x-amz-object-lock-retain-until-date", HeaderValue::from_static("2029-01-01T00:00:00Z"));
set_requested_object_lock_retention(&mut req);
let action = Action::S3Action(S3Action::PutObjectRetentionAction);
let conditions = remaining_retention_days_conditions(&req, action);
assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["COMPLIANCE".to_string()]));
let applied = OffsetDateTime::from(
req.input
.retention
.as_ref()
.expect("body retention")
.retain_until_date
.clone()
.expect("body date"),
);
let authorized = OffsetDateTime::parse(
&conditions["object-lock-retain-until-date"][0],
&time::format_description::well_known::Rfc3339,
)
.expect("condition date should parse");
assert_eq!(authorized.unix_timestamp(), applied.unix_timestamp());
let policy = Policy::parse_config(br#"{
"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:PutObjectRetention","Resource":"arn:aws:s3:::bucket/*",
"Condition":{"DateLessThan":{"s3:object-lock-retain-until-date":"2030-01-01T00:00:00Z"}}}]
}"#).expect("bounded retention policy should parse");
let args = Args {
account: "writer",
groups: &None,
action,
bucket: "bucket",
conditions: &conditions,
is_owner: false,
object: "object",
claims: &HashMap::new(),
deny_only: false,
};
assert!(
!policy.is_allowed(&args).await,
"an in-range header cannot authorize an out-of-range body"
);
req.input.retention = None;
set_requested_object_lock_retention(&mut req);
let conditions = remaining_retention_days_conditions(&req, action);
assert!(!conditions.contains_key("object-lock-mode"));
assert!(!conditions.contains_key("object-lock-retain-until-date"));
}
fn request_retaining_for(retain_for: time::Duration) -> S3Request<()> {
let mut req = build_request((), Method::PUT);
let retain_until = OffsetDateTime::now_utc() + retain_for;
+20 -12
View File
@@ -402,6 +402,25 @@ async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Re
const MAXIMUM_RETENTION_DAYS: i32 = 36_500;
const MAXIMUM_RETENTION_YEARS: i32 = 100;
pub(crate) fn object_lock_retention_from_metadata(metadata: &HashMap<String, String>) -> S3Result<ObjectLockRetention> {
let mode = metadata
.get(AMZ_OBJECT_LOCK_MODE_LOWER)
.map(|value| ObjectLockRetentionMode::from(value.as_str().to_string()));
let retain_until_date = metadata
.get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
.and_then(|value| OffsetDateTime::parse(value, &Rfc3339).ok())
// The S3 timestamp serializer emits a literal Z, so its wall clock
// must be UTC before it is serialized.
.map(|date| {
date.checked_to_offset(time::UtcOffset::UTC)
.filter(|date| (0..=9999).contains(&date.year()))
.map(Timestamp::from)
.ok_or_else(|| s3_error!(InternalError, "Stored retention date is outside the supported UTC range"))
})
.transpose()?;
Ok(ObjectLockRetention { mode, retain_until_date })
}
fn invalid_object_lock_configuration(message: impl Into<String>) -> S3Error {
S3Error::with_message(S3ErrorCode::MalformedXML, message.into())
}
@@ -1112,19 +1131,8 @@ impl S3 for FS {
s3_error!(InternalError, "{}", e.to_string())
})?;
let mode = object_info
.user_defined
.get(AMZ_OBJECT_LOCK_MODE_LOWER)
.map(|v| ObjectLockRetentionMode::from(v.as_str().to_string()));
let retain_until_date = object_info
.user_defined
.get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
.and_then(|v| OffsetDateTime::parse(v.as_str(), &Rfc3339).ok())
.map(Timestamp::from);
let output = GetObjectRetentionOutput {
retention: Some(ObjectLockRetention { mode, retain_until_date }),
retention: Some(object_lock_retention_from_metadata(&object_info.user_defined)?),
};
let version_id = req.input.version_id.clone().unwrap_or_default();
helper = helper.object(object_info).version_id(version_id);
+4
View File
@@ -340,6 +340,10 @@ pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> {
));
}
if key.is_empty() || !crate::storage::storage_api::ecstore_bucket::utils::is_valid_object_prefix(key) {
return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid object key"));
}
// Log debug info for keys with special characters to help diagnose encoding issues
if key.contains([' ', '+', '%']) {
debug!(
+40 -1
View File
@@ -18,7 +18,8 @@ mod tests {
use crate::server::cors;
use crate::storage::StorageError;
use crate::storage::ecfs::{
FS, parse_object_version_id, propagate_object_lock_peer_reload, validate_object_lock_configuration_input,
FS, object_lock_retention_from_metadata, parse_object_version_id, propagate_object_lock_peer_reload,
validate_object_lock_configuration_input,
};
use crate::storage::ecfs_extend::{apply_bucket_default_lock_retention, map_bucket_object_lock_config_state};
use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
@@ -988,6 +989,44 @@ mod tests {
assert!(validate_list_object_unordered_with_delimiter(delimiter_some, complex_query_without_unordered).is_ok());
}
#[test]
fn retention_response_serialization_preserves_offset_metadata_instant() {
for value in [
"2035-01-02T00:00:00+23:59",
"2035-01-02T00:00:00-23:59",
"2035-01-02T00:00:00Z",
] {
let metadata = HashMap::from([
(AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
]);
let retention = object_lock_retention_from_metadata(&metadata).expect("representable retention instant");
assert_eq!(retention.mode.as_ref().expect("mode").as_str(), "COMPLIANCE");
let mut serialized = Vec::new();
retention
.retain_until_date
.expect("stored retention date")
.format(s3s::dto::TimestampFormat::DateTime, &mut serialized)
.expect("serialize the response timestamp");
let serialized = String::from_utf8(serialized).expect("UTF-8 timestamp");
let returned = OffsetDateTime::parse(&serialized, &Rfc3339).expect("parse wire timestamp");
let stored = OffsetDateTime::parse(value, &Rfc3339).expect("parse stored timestamp");
assert_eq!(returned.unix_timestamp(), stored.unix_timestamp(), "stored date {value}");
}
}
#[test]
fn retention_response_rejects_instants_outside_the_utc_wire_range() {
for value in ["9999-12-31T23:59:59-23:59", "0000-01-01T00:00:00+23:59"] {
let metadata = HashMap::from([
(AMZ_OBJECT_LOCK_MODE_LOWER.to_owned(), "COMPLIANCE".to_owned()),
(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_owned(), value.to_owned()),
]);
let error = object_lock_retention_from_metadata(&metadata).expect_err("unrepresentable UTC timestamp");
assert_eq!(error.code(), &S3ErrorCode::InternalError);
}
}
#[test]
fn test_parse_object_lock_retention() {
use time::macros::datetime;
+41 -4
View File
@@ -368,6 +368,13 @@ pub async fn put_opts_with_replication_authorization(
metadata: HashMap<String, String>,
replication_request_authorized: bool,
) -> Result<ObjectOptions> {
if vid.is_some() && !replication_request_authorized {
return Err(StorageError::InvalidArgument(
bucket.to_owned(),
object.to_owned(),
"A destination versionId requires replication authorization".to_owned(),
));
}
let versioning_cfg = bucket_versioning_config_for_write(bucket).await?;
let versioned = versioning_cfg.prefix_enabled(object);
let version_suspended = versioning_cfg.prefix_suspended(object);
@@ -1532,7 +1539,15 @@ mod tests {
let headers = create_test_headers();
let invalid_uuid = "invalid-uuid".to_string();
let result = put_opts("test-bucket", "test-object", Some(invalid_uuid), &headers, HashMap::new()).await;
let result = put_opts_with_replication_authorization(
"test-bucket",
"test-object",
Some(invalid_uuid),
&headers,
HashMap::new(),
true,
)
.await;
assert!(result.is_err());
if let Err(err) = result {
@@ -1556,13 +1571,35 @@ mod tests {
// get_opts / del_opts already do.
let headers = create_test_headers();
let opts = put_opts("test-bucket", "test-object", Some("null".to_string()), &headers, HashMap::new())
.await
.expect("PUT with versionId=null must be accepted as the null version");
let opts = put_opts_with_replication_authorization(
"test-bucket",
"test-object",
Some("null".to_string()),
&headers,
HashMap::new(),
true,
)
.await
.expect("PUT with versionId=null must be accepted as the null version");
assert_eq!(opts.version_id, Some(Uuid::nil().to_string()));
}
#[tokio::test]
async fn ordinary_writes_cannot_choose_a_destination_version() {
let headers = create_test_headers();
for version in ["null", "", "0194e0f1-0000-7000-8000-000000000000"] {
let err = put_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
.await
.expect_err("ordinary PUT must not overwrite a selected version");
assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
let err = copy_dst_opts("test-bucket", "test-object", Some(version.to_owned()), &headers, HashMap::new())
.await
.expect_err("ordinary COPY must not overwrite a selected version");
assert!(matches!(err, StorageError::InvalidArgument(_, _, _)));
}
}
#[tokio::test]
async fn test_copy_dst_opts() {
let headers = create_test_headers();
+87 -2
View File
@@ -226,7 +226,11 @@ pub(crate) fn build_list_object_versions_output(
size: Some(v.size),
version_id: Some(list_versions_response_version_id(v.version_id)),
is_latest: Some(v.is_latest),
e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
e_tag: v
.etag
.as_ref()
.filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
.map(|etag| to_s3s_etag(etag)),
storage_class: v.storage_class.clone().map(ObjectVersionStorageClass::from),
..Default::default()
})
@@ -311,7 +315,11 @@ pub(crate) fn build_list_objects_v2_output(
// logical-size sentinel; never expose that internal value in
// an S3 response.
size: Some(v.get_actual_size_or_physical()),
e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
e_tag: v
.etag
.as_ref()
.filter(|_| rustfs_utils::http::etag_can_be_listed(&v.user_defined))
.map(|etag| to_s3s_etag(etag)),
storage_class: v.storage_class.clone().map(ObjectStorageClass::from),
..Default::default()
};
@@ -709,6 +717,83 @@ mod tests {
);
}
#[test]
fn standard_list_responses_hide_legacy_encrypted_fingerprints() {
let mut protected_metadata = std::collections::HashMap::from([(
"x-amz-server-side-encryption-customer-algorithm".to_owned(),
"AES256".to_owned(),
)]);
rustfs_utils::http::metadata_compat::insert_str(
&mut protected_metadata,
rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
"v1".to_owned(),
);
let objects: Vec<ObjectInfo> = [
std::collections::HashMap::from([(
"x-amz-server-side-encryption-customer-algorithm".to_owned(),
"AES256".to_owned(),
)]),
std::collections::HashMap::from([("x-amz-server-side-encryption".to_owned(), "aws:kms".to_owned())]),
protected_metadata,
std::collections::HashMap::new(),
]
.into_iter()
.enumerate()
.map(|(index, metadata)| ObjectInfo {
name: format!("object-{index}"),
etag: Some("0123456789abcdef0123456789abcdef".to_owned()),
user_defined: std::sync::Arc::new(metadata),
size: 42,
..Default::default()
})
.collect();
let v2 = build_list_objects_v2_output(
ListObjectsV2Info {
objects: objects.clone(),
..Default::default()
},
false,
1000,
"bucket".to_owned(),
String::new(),
None,
None,
None,
None,
);
let v1 = build_list_objects_output(v2.clone(), None);
for listing in [&v1.contents, &v2.contents] {
let entries = listing.as_ref().expect("listed objects");
assert_eq!(entries.len(), 4);
for (index, entry) in entries.iter().enumerate() {
assert_eq!(entry.e_tag.is_some(), index >= 2);
assert_eq!(entry.key.as_deref(), Some(format!("object-{index}").as_str()));
assert_eq!(entry.size, Some(42));
}
}
let versions = build_list_object_versions_output(
ListObjectVersionsInfo {
objects,
..Default::default()
},
"bucket".to_owned(),
&ListObjectVersionsParams {
prefix: String::new(),
delimiter: None,
key_marker: None,
version_id_marker: None,
max_keys: 1000,
},
None,
);
let versions = versions.versions.expect("listed versions");
assert_eq!(versions.len(), 4);
for (index, entry) in versions.iter().enumerate() {
assert_eq!(entry.e_tag.is_some(), index >= 2);
assert_eq!(entry.size, Some(42));
}
}
#[test]
fn list_responses_report_standard_for_legacy_label_only_file_metadata() {
let version_id = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("fixture version ID should be valid");
+1
View File
@@ -4127,6 +4127,7 @@ pub fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>)
for key in KEYS.iter() {
metadata.remove(*key);
}
rustfs_utils::http::metadata_compat::remove_str(metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
}
pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap<String, String>) {
+154
View File
@@ -0,0 +1,154 @@
"""Verify the real locked capture graph and leader identity feature contract."""
import hashlib
import json
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import re
import unittest
ROOT = Path(__file__).resolve().parents[2]
REGISTRY = "registry+https://github.com/rust-lang/crates.io-index"
def source_files(root):
result = subprocess.run(["git", "ls-files", "-z"], cwd=root, check=True, capture_output=True)
paths = {Path(os.fsdecode(p)) for p in result.stdout.split(b"\0") if p}
paths.add(Path("scripts/tests/test_capture_dependencies.py"))
return sorted(paths)
def fingerprint(root, paths):
digest = hashlib.sha256()
for path in paths:
digest.update(os.fsencode(path))
digest.update(b"\0")
source = root / path
if source.is_symlink():
digest.update(b"symlink\0")
digest.update(os.fsencode(os.readlink(source)))
else:
digest.update(b"file\0")
digest.update(source.read_bytes())
return digest.hexdigest()
def metadata(root, all_features=False):
lock = (root / "Cargo.lock").read_bytes()
command = ["cargo", "metadata", "--locked", "--format-version", "1"]
if all_features:
command.append("--all-features")
result = subprocess.run(command, cwd=root, capture_output=True, text=True)
if result.returncode:
raise RuntimeError(f"{' '.join(command)} exited {result.returncode}: {result.stderr}")
if (root / "Cargo.lock").read_bytes() != lock:
raise AssertionError("CAPTURE_LOCK_CHANGED")
print(f"{' '.join(command)}: exit 0; lock {hashlib.sha256(lock).hexdigest()}")
return json.loads(result.stdout)
def assert_graph(case, root, graph):
packages = {package["id"]: package for package in graph["packages"]}
capture = next(p for p in packages.values() if p["name"] == "rustfs-capture-authority")
dependencies = {p["name"]: p for p in capture["dependencies"]}
case.assertIn("openraft", dependencies, "CAPTURE_OPENRAFT_REQUIRED")
dependency = dependencies["openraft"]
case.assertEqual(dependency["req"], "=0.9.25", "CAPTURE_OPENRAFT_EXACT_PIN")
case.assertFalse(dependency["uses_default_features"], "CAPTURE_OPENRAFT_DEFAULTS_DISABLED")
case.assertIn("serde", dependency["features"])
case.assertEqual(dependency["source"], REGISTRY)
for name, kind in (("serde", None), ("url", None), ("serde_json", "dev")):
case.assertIn(name, dependencies)
case.assertEqual(dependencies[name]["kind"], kind)
case.assertIn("derive", dependencies["serde"]["features"])
nodes = {node["id"]: node for node in graph["resolve"]["nodes"]}
reachable = set()
pending = [capture["id"]]
while pending:
node_id = pending.pop()
if node_id not in reachable:
reachable.add(node_id)
pending.extend(nodes[node_id]["dependencies"])
openraft = [p for p in packages.values() if p["name"] == "openraft"]
case.assertEqual(len(openraft), 1, "CAPTURE_UNIQUE_OPENRAFT")
package = openraft[0]
case.assertIn(package["id"], reachable)
case.assertEqual(package["version"], "0.9.25")
case.assertEqual(package["source"], REGISTRY)
features = set(nodes[package["id"]]["features"])
case.assertNotIn("single-term-leader", features, "CAPTURE_ADVANCED_LEADER_REQUIRED")
case.assertNotIn("storage-v2", features, "CAPTURE_STORAGE_V1_REQUIRED")
case.assertIn("serde", features)
case.assertEqual(set(dependency["features"]), {"serde"}, "CAPTURE_OPENRAFT_SERDE_ONLY")
for name in ("serde", "serde_derive", "url", "serde_json"):
case.assertTrue(any(packages[p]["name"] == name for p in reachable), name)
case.assertTrue(any("derive" in nodes[p]["features"] for p in reachable if packages[p]["name"] == "serde"))
blocks = (root / "Cargo.lock").read_text().split("[[package]]")
locked = [dict(re.findall(r'^([a-z]+) = "([^"\n]*)"$', block, re.MULTILINE))
for block in blocks if re.search(r'^name = "openraft"$', block, re.MULTILINE)]
case.assertEqual(len(locked), 1)
case.assertEqual(locked[0]["version"], "0.9.25")
case.assertEqual(locked[0]["source"], REGISTRY)
case.assertEqual(locked[0]["checksum"], "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd", "CAPTURE_PUBLISHED_OPENRAFT_CHECKSUM")
class CaptureDependencies(unittest.TestCase):
def test_accepted_dependency_graph(self):
for all_features in (False, True):
with self.subTest(all_features=all_features):
graph = metadata(ROOT, all_features)
assert_graph(self, ROOT, graph)
package = next(p for p in graph["packages"] if p["name"] == "openraft")
node = next(n for n in graph["resolve"]["nodes"] if n["id"] == package["id"])
print(f"accepted all_features={all_features}: {package['id']}; features {node['features']}")
def control(self, old, new, expected):
paths = source_files(ROOT)
before = fingerprint(ROOT, paths)
with tempfile.TemporaryDirectory(prefix="capture-dependencies-") as directory:
copied = Path(directory)
for path in paths:
destination = copied / path
destination.parent.mkdir(parents=True, exist_ok=True)
source = ROOT / path
if source.is_symlink():
destination.symlink_to(os.readlink(source))
else:
shutil.copyfile(source, destination)
self.assertEqual(fingerprint(copied, paths), before)
manifest = copied / "Cargo.toml"
original = manifest.read_bytes()
self.assertEqual(original.count(old.encode()), 1)
try:
manifest.write_bytes(original.replace(old.encode(), new.encode(), 1))
mutated = fingerprint(copied, paths)
self.assertNotEqual(mutated, before)
for all_features in (False, True):
graph = metadata(copied, all_features)
versions = {p["version"] for p in graph["packages"] if p["name"] == "openraft"}
self.assertEqual(versions, {"0.9.25"})
with self.assertRaisesRegex(AssertionError, expected):
assert_graph(self, copied, graph)
print(f"{expected}: metadata exit 0 (default/all-features); source {before}; control {mutated}")
finally:
manifest.write_bytes(original)
self.assertEqual(fingerprint(copied, paths), before)
self.assertEqual(fingerprint(ROOT, paths), before)
print(f"{expected}: restored source {before}")
def test_relaxed_pin_is_rejected(self):
self.control('version = "=0.9.25"', 'version = "0.9.25"', "CAPTURE_OPENRAFT_EXACT_PIN")
def test_single_term_leader_is_rejected(self):
self.control(
'openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }',
'openraft = { version = "=0.9.25", default-features = false, features = ["serde", "single-term-leader"] }',
"CAPTURE_ADVANCED_LEADER_REQUIRED",
)
if __name__ == "__main__":
unittest.main()