gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 02:56:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(replication): split oversized hot-path functions, proxy unreplicated reads, and fail SSE-C passthrough closed (#6170)

Browse Source 
* refactor(replication): split four oversized hot-path functions into focused helpers

Pure-move decomposition of the four oversized functions flagged by the
replication compatibility review (P1-18), unblocking migration milestone
M2 which requires resyncer moves to stay mechanical:

- resync_bucket (522 lines -> 61-line step sequence): leader lock,
  target resolution, walk/collector/worker spawning, and dispatch loop
  extracted into focused helpers; pure decision helpers (DTO builders,
  HEAD-result classification) separated from IO orchestration.
- replicate_all (411 lines -> 113-line main body): initial target-info
  seeding, read/stat option builders, skip-path notes, target HEAD
  action resolution, and the multipart/single-put payload transport
  extracted as private free functions.
- start_mrf_processor (306 lines -> 46-line spawn body): recovery guard,
  ledger load, per-entry replay (delete/object/metadata), and retained
  entry resolution extracted; retry bookkeeping semantics preserved
  exactly (inner continue-paths push inside helpers, outer Missed push
  stays in the loop).
- apply_iam_item (255 lines -> match dispatch skeleton): one helper per
  IAM item type.

No behavior change: log texts, error paths, event emissions, and metric
counts are byte-identical; existing tests unchanged and green (238
ecstore replication/mrf/resync + 232 rustfs site-replication).

* feat(replication): proxy GET/HEAD/Tagging for unreplicated objects to replication targets (#6172)

* feat(replication): proxy GET/HEAD/Tagging for unreplicated objects to replication targets

Implements the MinIO active-active read-proxy protocol (P1-5 of the
replication compatibility review): when a GET/HEAD/GetObjectTagging/
PutObjectTagging/DeleteObjectTagging request fails locally with
not-found and the bucket has replication targets, the request is proxied
to the targets in rule order, mirroring bucket-replication.go
proxyGetToReplicationTarget/proxyHeadToRepTarget/proxyTaggingToRepTarget.

Protocol surface:
- Anti-loop: inbound {x-rustfs-,x-minio-}source-proxy-request is parsed
  into ObjectOptions (proxy_request + proxy_header_set, matching MinIO
  ProxyRequest/ProxyHeaderSet); a request carrying the marker with ANY
  value is never re-proxied. Outbound client proxy calls send the marker
  as "true"; replication worker convergence HEADs send it as "false" so
  a peer's proxy layer cannot answer a convergence check by proxying
  back to the source (which would fake Completed without a PUT).
- Target selection: new replication_proxy.rs get_proxy_targets — empty
  when the marker is set, versioning is suspended, or no replication
  config; otherwise filter_target_arns -> TargetClient lookup, skipping
  targets with proxying disabled.
- TargetClient gains head_object_for_proxy/get_object (streaming) and
  the three tagging calls. Proxy calls never send the replication-check
  SSE-C exemption header; customer SSE-C keys are forwarded verbatim so
  the target performs real decryption. Conditional (If-*) headers are
  not forwarded (MinIO parity); Range and part_number are, with
  parts_count/tag_count/storage_class/expiration passed through.
- Metrics: proxy counters now count only real client proxy traffic,
  MinIO-aligned (one total per proxied request, one failed when no
  target served it). The previous misattributed counters — replication
  worker HEAD/PUT (#2672) and local tagging operations (#2682) — are
  removed; ReplProxyMetric now maps the tagging counters instead of
  dropping them.

e2e (fake_s3_target extended with tagging + header journaling): proxied
GET body + outbound header contract (marker present, no
replication-check, SSE-C passthrough), HEAD, anti-loop 404 with zero
outbound requests, GetObjectTagging, and metric mapping unit tests.

Rolling note: proxying only activates for buckets with replication
targets; requests carrying the marker keep pre-upgrade behavior.

Refs rustfs/backlog#1675 (P1-5)

* fix(replication): fail SSE-C passthrough closed on targets that drop transport headers (#6178)

SSE-C ciphertext passthrough replicates via X-Rustfs-Replication-* transport
headers. A MinIO/generic-S3 target silently discards them, storing bare
ciphertext with no decryption material — yet the PUT succeeded, so the object
reported COMPLETED with a silently unreadable replica (backlog#1675 N2).

Fail-closed design:
- SsecPassthroughCapability {Unknown, Supported, Unsupported} cached in
  BucketTargetSys per target ARN with a recording timestamp. Entries reset
  whenever the target is rebuilt, edited, or removed (arn_remotes_map
  lifecycle) and expire after SSEC_PASSTHROUGH_CAPABILITY_TTL (10 minutes):
  an expired verdict in either direction is re-earned through the audit, so
  an Unsupported target recovers automatically after an upgrade (at most one
  wasted PUT+HEAD audit per bad target per TTL window) and a Supported
  verdict cannot outlive a backend swapped behind the same endpoint.
- Replication worker (replicate_object and replicate_all): fresh Unsupported
  targets never receive the PUT — the attempt fails immediately into the
  normal MRF retry channel with a "run ?replication-check to re-probe" hint.
  Unknown or expired verdicts are audited: after the PUT the worker HEADs
  the replica back through the replication-check channel (source version id
  mapped through resolve_read_api_version_id, so null-version objects audit
  correctly) and requires SSE-C evidence (the echoed customer-algorithm
  header); missing evidence records Unsupported and fails the attempt.
  Convergence HEADs are audited the same way, so a broken ciphertext replica
  from an earlier attempt can never launder itself into COMPLETED via an
  ETag match. The gate/evidence policy is pure (replication_target_boundary,
  staleness folded in as an input) for the M2 worker migration.
- replication-check grows an SsecPassthrough probe phase: a probe PUT
  carrying the live transport-header shape, HEAD-back for evidence, and a
  machine-readable Code BucketRemoteSsecPassthroughUnsupported on failure.
  The probe verdict is synced into the runtime capability cache. Unlike
  VersionFidelity, a failed SsecPassthrough phase does NOT fail the target
  overall — it is a capability limit, not a broken replication contract,
  and a plaintext-only deployment against such a target must not turn red.
- fake_s3_target: default mode now models a RustFS target (stores the
  transport headers, echoes SSE-C evidence); the new
  drop_unlisted_replication_headers mode models MinIO. The journal records
  whether a request carried transport headers.

Receiver-echo verification: the replication-check HEAD exemption only skips
SSE-C key validation; the response has always built sse-customer-algorithm
from stored metadata (rustfs/src/app/object_usecase.rs), so no receiver
change was needed — pinned end to end by the replication-check e2e against
a real RustFS target.

Rolling-upgrade constraint: RustFS targets older than the replication-check
HEAD exemption (#5898) answer the audit HEAD without SSE-C evidence (or fail
it outright), so SSE-C replication to such targets reports FAILED. This is
deliberate — FAILED-and-retryable beats a silently undecryptable replica —
and self-heals: once the target is upgraded, the next TTL expiry (or a
manual ?replication-check re-probe) re-audits and records Supported.
Plaintext and managed-SSE replication are unaffected. The capability cache
is per-node; each node audits independently.

Known limitations:
- The audit judges evidence from the echoed customer-algorithm header only.
  A hypothetical target that preserves that one header while dropping other
  transport headers (partial-drop) would pass the audit; no known target
  behaves this way — observed targets drop the whole unknown-header family.
- A mixed-version target cluster can flap the verdict between audits routed
  to different target nodes until the rollout completes; the TTL bounds how
  long each stale verdict persists.

New e2e (backlog#1675 C1 + N2, red-first): fail-closed against a
header-dropping fake (FAILED + no second PUT via the capability cache,
journal-asserted; red run showed the old COMPLETED), replication-check
reports the SsecPassthrough phase Code while the target stays OK overall,
SSE-C heal convergence after a real target outage, and SSE-C
existing-object resync landing a REPLICA readable with the customer key.
TTL expiry in both directions is pinned at the cache and gate seams.

* refactor(replication): move resyncer pure decision logic into rustfs-replication (M2) (#6180)

* refactor(replication): move resyncer pure decision logic into rustfs-replication (M2)

Pure-move milestone M2 of the ECStore replication split (backlog#1675
P1-17): relocate the resyncer's IO-free decision helpers, with their unit
tests, into the crates they already belong to by type ownership. No
behavior change.

Moved into crates/replication:
- resync.rs: resync_status_duration
- delete.rs: resync_existing_delete_replication_info,
  replicate_delete_outcome, target_delete_version_id,
  delete_marker_purge_version_id, delete_marker_purge_mrf_entry
- object.rs: version_identity_drifted, is_replication_target_offline_error,
  SsecPassthroughCapability, SsecPassthroughGate, ssec_passthrough_gate,
  ssec_passthrough_evidence_present (param-demoted to the echoed
  customer-algorithm string; ECStore keeps the HeadObjectOutput adapter)
- filemeta.rs: NULL_VERSION_ID wire literal (crate-owned copy per the
  filemeta-independence contract)

ECStore rewiring (Rule #14: imports stay in *_boundary.rs):
- resync/object-decision/target boundaries re-export the moved symbols;
  resyncer call sites are unchanged
- bucket_target_sys keeps only the verdict cache + TTL and re-exports the
  capability enum so existing consumer paths keep compiling

Not moved (signatures carry ECStore or aws-sdk types):
verify_resync_head_result, resync_target_error_detail, the SdkError
classifiers, the replicate_all_* option/info builders, and the env-coupled
bounded_resync_max_jobs admission clamp. README milestone table updated.

* chore(replication): retire the datatypes.rs relay early

README sanctions retiring datatypes.rs ahead of M4. The module was a
pure relay (resync boundary -> datatypes -> mod.rs facade) with no
external consumer importing it directly, so the facade now re-exports
ResyncStatusType from replication_resync_boundary and the relay file is
deleted. Consumers stay behind the ECStore facade, keeping Migration
Rule #15 intact — the original retirement wording ("consumers import
through rustfs-replication directly") conflicted with that rule and is
corrected in the README.

* chore(arch): extend migration guards to the M2-moved decision contracts

The adversarial review of the M2 move found the per-symbol ratchet in
check_architecture_migration_rules.sh was not extended for the moved
symbols, leaving them free to be redefined in ECStore or imported past
their boundary without CI noticing:

- resync definition pin + boundary fences gain resync_status_duration;
- the object-decision boundary fences gain the five delete-family
  helpers (delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
  replicate_delete_outcome, resync_existing_delete_replication_info,
  target_delete_version_id);
- the target-boundary fence gains the SSE-C gate family, the offline
  classifier, and version_identity_drifted;
- a new definition pin rejects ECStore redefinitions of the M2-moved
  fns/enums (ssec_passthrough_evidence_present deliberately excluded:
  ECStore keeps a thin HeadObjectOutput adapter under that name).

Mutation-verified: a probe fn ssec_passthrough_gate under
crates/ecstore/src/bucket/replication trips the new pin.

Also anchors the intentionally-duplicated NULL_VERSION_ID wire literal
from the filemeta side and tightens the M2 README note on
bounded_resync_max_jobs.
This commit is contained in:
唐小鸭
2026-08-18 21:45:38 +08:00
committed by GitHub
parent 27d23b6135
commit 1cf0f7af15
33 changed files with 4694 additions and 1612 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/site_replication.rs
+262 -242
View File
@@ -66,18 +66,20 @@ use rustfs_config::{
};
use rustfs_iam::error::is_err_no_such_service_account;
use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
use rustfs_iam::store::object::ObjectStore;
use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type, user_type_from_sr_wire};
use rustfs_iam::sys::{
NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
IamSys, NewServiceAccountOpts, SITE_REPLICATOR_SERVICE_ACCOUNT, UpdateServiceAccountOpts, get_claims_from_token_with_secret,
};
use rustfs_madmin::{
AddOrUpdateUserReq, BucketBandwidth, GroupAddRemove, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric,
LDAPConfigSettings, LDAPSettings, OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus,
ReplicateEditStatus, ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC,
SR_IAM_ITEM_STS_ACC_LEGACY, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem,
SRIAMPolicy, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation,
SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRSiteSummary,
SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
SRIAMPolicy, SRIAMUser, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq,
SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRRetryStats, SRSTSCredential,
SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate,
SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
};
use rustfs_policy::policy::{
Policy,
@@ -4584,7 +4586,10 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
head_failed_total: non_negative_u64(node.proxy_head_failed),
put_tag_total: non_negative_u64(node.proxy_put_tag_total),
put_tag_failed_total: non_negative_u64(node.proxy_put_tag_failed),
..Default::default()
get_tag_total: non_negative_u64(node.proxy_get_tag_total),
get_tag_failed_total: non_negative_u64(node.proxy_get_tag_failed),
remove_tag_total: non_negative_u64(node.proxy_delete_tag_total),
remove_tag_failed_total: non_negative_u64(node.proxy_delete_tag_failed),
},
metrics,
uptime: node.uptime,
@@ -9358,247 +9363,16 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
let incoming_updated_at = item.updated_at;
match item.r#type.as_str() {
"policy" => {
if let Some(policy) = item.policy {
let policy: Policy =
serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
iam_sys.set_policy(&item.name, policy).await.map_err(ApiError::from)?;
} else {
iam_sys.delete_policy(&item.name, true).await.map_err(ApiError::from)?;
}
Ok(())
}
"policy-mapping" => {
let Some(mapping) = item.policy_mapping else {
return Err(s3_error!(InvalidRequest, "policyMapping is required"));
};
let user_type =
user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
iam_sys
.policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
"group-info" => {
let Some(group_info) = item.group_info else {
return Err(s3_error!(InvalidRequest, "groupInfo is required"));
};
let update = group_info.update_req;
if !group_info_requires_upsert(&update) {
iam_sys
.remove_users_from_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
return Ok(());
}
iam_sys
.add_users_to_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
iam_sys
.set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
.await
.map_err(ApiError::from)?;
Ok(())
}
"policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy).await,
"policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping).await,
"group-info" => apply_iam_group_info_item(&iam_sys, item.group_info).await,
// MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias
// `sts-credential` (emitted by older RustFS releases) stays accepted permanently
// so mixed-version RustFS sites keep replicating STS credentials during rolling
// upgrades; it is a compatibility layer, not temporary code.
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => {
let Some(sts_credential) = item.sts_credential else {
return Err(s3_error!(InvalidRequest, "stsCredential is required"));
};
let Some(secret) = current_token_signing_key() else {
return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
};
let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
.map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
let expiration = claims
.get("exp")
.and_then(claims_unix_timestamp)
.map(OffsetDateTime::from_unix_timestamp)
.transpose()
.map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
let groups = string_list_claim(&claims, "groups");
let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
let cred = rustfs_credentials::Credentials {
access_key: sts_credential.access_key.clone(),
secret_key: sts_credential.secret_key.clone(),
session_token: sts_credential.session_token.clone(),
expiration,
status: "on".to_string(),
parent_user: sts_credential.parent_user.clone(),
groups,
claims: Some(claims),
..Default::default()
};
iam_sys
.set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
"iam-user" => {
let Some(user) = item.iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
} else {
let Some(user_req) = user.user_req else {
return Err(s3_error!(InvalidRequest, "userReq is required"));
};
let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
if is_status_only_update {
iam_sys
.set_user_status(&user.access_key, user_req.status)
.await
.map_err(ApiError::from)?;
} else {
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
}
}
Ok(())
}
"service-account" => {
let Some(change) = item.svc_acc_change else {
return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
};
let envelope = change.oidc_service_account_envelope;
if let Some(create) = change.create {
let local_updated_at = iam_sys
.get_user(&create.access_key)
.await
.map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
return Ok(());
}
ReplicatedServiceAccountPolicy {
policy: Some(site_replicator_service_account_policy()?),
is_envelope: false,
}
} else {
let Some(replicated_policy) = decode_service_account_replication_policy(
&create,
envelope.as_ref(),
incoming_updated_at,
local_updated_at,
)?
else {
return Ok(());
};
replicated_policy
};
match iam_sys.get_service_account(&create.access_key).await {
Ok((existing, _)) => {
if existing.parent_user != create.parent {
return Err(s3_error!(
InvalidRequest,
"service account {} already exists with a different parent user",
create.access_key
));
}
iam_sys
.update_service_account(
&create.access_key,
UpdateServiceAccountOpts {
name: replicated_policy.metadata_for_existing_account(create.name),
description: replicated_policy.metadata_for_existing_account(create.description),
session_policy: replicated_policy.for_existing_account(),
secret_key: Some(create.secret_key),
expiration: create.expiration,
status: (!create.status.is_empty()).then_some(create.status),
parent_user: None,
allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) if is_err_no_such_service_account(&err) => {
iam_sys
.new_service_account(
&create.parent,
Some(create.groups),
NewServiceAccountOpts {
session_policy: replicated_policy.policy,
access_key: create.access_key,
secret_key: create.secret_key,
name: (!create.name.is_empty()).then_some(create.name),
description: (!create.description.is_empty()).then_some(create.description),
expiration: create.expiration,
allow_site_replicator_account: true,
claims: Some(create.claims),
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) => return Err(ApiError::from(err).into()),
}
return Ok(());
}
if let Some(update) = change.update {
if let Some(local) = iam_sys.get_user(&update.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
let session_policy = if allow_site_replicator_account {
Some(site_replicator_service_account_policy()?)
} else {
update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
};
iam_sys
.update_service_account(
&update.access_key,
UpdateServiceAccountOpts {
session_policy,
secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
name: (!update.name.is_empty()).then_some(update.name),
description: (!update.description.is_empty()).then_some(update.description),
expiration: update.expiration,
status: (!update.status.is_empty()).then_some(update.status),
// Peers replicate credentials, never the local parent binding:
// each site resolves its own parent from its own IAM.
parent_user: None,
allow_site_replicator_account,
},
)
.await
.map_err(ApiError::from)?;
return Ok(());
}
if let Some(delete) = change.delete {
if let Some(local) = iam_sys.get_user(&delete.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
iam_sys
.delete_service_account(&delete.access_key, true)
.await
.map_err(ApiError::from)?;
return Ok(());
}
Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
}
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => apply_iam_sts_account_item(&iam_sys, item.sts_credential).await,
"iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await,
"service-account" => apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await,
_ => Err(s3_error!(
NotImplemented,
"site replication IAM item type `{}` is not supported",
@@ -9607,6 +9381,252 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
}
}
async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, policy: Option<Value>) -> S3Result<()> {
if let Some(policy) = policy {
let policy: Policy =
serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?;
} else {
iam_sys.delete_policy(name, true).await.map_err(ApiError::from)?;
}
Ok(())
}
async fn apply_iam_policy_mapping_item(iam_sys: &IamSys<ObjectStore>, policy_mapping: Option<SRPolicyMapping>) -> S3Result<()> {
let Some(mapping) = policy_mapping else {
return Err(s3_error!(InvalidRequest, "policyMapping is required"));
};
let user_type = user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
iam_sys
.policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Option<SRGroupInfo>) -> S3Result<()> {
let Some(group_info) = group_info else {
return Err(s3_error!(InvalidRequest, "groupInfo is required"));
};
let update = group_info.update_req;
if !group_info_requires_upsert(&update) {
iam_sys
.remove_users_from_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
return Ok(());
}
iam_sys
.add_users_to_group(&update.group, update.members)
.await
.map_err(ApiError::from)?;
iam_sys
.set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_sts_account_item(iam_sys: &IamSys<ObjectStore>, sts_credential: Option<SRSTSCredential>) -> S3Result<()> {
let Some(sts_credential) = sts_credential else {
return Err(s3_error!(InvalidRequest, "stsCredential is required"));
};
let Some(secret) = current_token_signing_key() else {
return Err(s3_error!(InvalidRequest, "token signing key not initialized"));
};
let claims = get_claims_from_token_with_secret(&sts_credential.session_token, &secret)
.map_err(|e| s3_error!(InvalidRequest, "invalid STS session token: {e}"))?;
let expiration = claims
.get("exp")
.and_then(claims_unix_timestamp)
.map(OffsetDateTime::from_unix_timestamp)
.transpose()
.map_err(|e| s3_error!(InvalidRequest, "invalid STS expiry: {e}"))?;
let groups = string_list_claim(&claims, "groups");
let compatibility_policy = sts_replication_compatibility_policy(&claims, &sts_credential.parent_policy_mapping);
let cred = rustfs_credentials::Credentials {
access_key: sts_credential.access_key.clone(),
secret_key: sts_credential.secret_key.clone(),
session_token: sts_credential.session_token.clone(),
expiration,
status: "on".to_string(),
parent_user: sts_credential.parent_user.clone(),
groups,
claims: Some(claims),
..Default::default()
};
iam_sys
.set_temp_user(&sts_credential.access_key, &cred, compatibility_policy)
.await
.map_err(ApiError::from)?;
Ok(())
}
async fn apply_iam_user_item(
iam_sys: &IamSys<ObjectStore>,
iam_user: Option<SRIAMUser>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<()> {
let Some(user) = iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
} else {
let Some(user_req) = user.user_req else {
return Err(s3_error!(InvalidRequest, "userReq is required"));
};
let is_status_only_update = user_req.secret_key.is_empty() && user_req.policy.is_none();
if is_status_only_update {
iam_sys
.set_user_status(&user.access_key, user_req.status)
.await
.map_err(ApiError::from)?;
} else {
iam_sys
.create_user(&user.access_key, &user_req)
.await
.map_err(ApiError::from)?;
}
}
Ok(())
}
async fn apply_iam_service_account_item(
iam_sys: &IamSys<ObjectStore>,
svc_acc_change: Option<SRSvcAccChange>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<()> {
let Some(change) = svc_acc_change else {
return Err(s3_error!(InvalidRequest, "serviceAccountChange is required"));
};
let envelope = change.oidc_service_account_envelope;
if let Some(create) = change.create {
let local_updated_at = iam_sys
.get_user(&create.access_key)
.await
.map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
return Ok(());
}
ReplicatedServiceAccountPolicy {
policy: Some(site_replicator_service_account_policy()?),
is_envelope: false,
}
} else {
let Some(replicated_policy) =
decode_service_account_replication_policy(&create, envelope.as_ref(), incoming_updated_at, local_updated_at)?
else {
return Ok(());
};
replicated_policy
};
match iam_sys.get_service_account(&create.access_key).await {
Ok((existing, _)) => {
if existing.parent_user != create.parent {
return Err(s3_error!(
InvalidRequest,
"service account {} already exists with a different parent user",
create.access_key
));
}
iam_sys
.update_service_account(
&create.access_key,
UpdateServiceAccountOpts {
name: replicated_policy.metadata_for_existing_account(create.name),
description: replicated_policy.metadata_for_existing_account(create.description),
session_policy: replicated_policy.for_existing_account(),
secret_key: Some(create.secret_key),
expiration: create.expiration,
status: (!create.status.is_empty()).then_some(create.status),
parent_user: None,
allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT,
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) if is_err_no_such_service_account(&err) => {
iam_sys
.new_service_account(
&create.parent,
Some(create.groups),
NewServiceAccountOpts {
session_policy: replicated_policy.policy,
access_key: create.access_key,
secret_key: create.secret_key,
name: (!create.name.is_empty()).then_some(create.name),
description: (!create.description.is_empty()).then_some(create.description),
expiration: create.expiration,
allow_site_replicator_account: true,
claims: Some(create.claims),
},
)
.await
.map_err(ApiError::from)?;
}
Err(err) => return Err(ApiError::from(err).into()),
}
return Ok(());
}
if let Some(update) = change.update {
if let Some(local) = iam_sys.get_user(&update.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
let allow_site_replicator_account = update.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT;
let session_policy = if allow_site_replicator_account {
Some(site_replicator_service_account_policy()?)
} else {
update.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok())
};
iam_sys
.update_service_account(
&update.access_key,
UpdateServiceAccountOpts {
session_policy,
secret_key: (!update.secret_key.is_empty()).then_some(update.secret_key),
name: (!update.name.is_empty()).then_some(update.name),
description: (!update.description.is_empty()).then_some(update.description),
expiration: update.expiration,
status: (!update.status.is_empty()).then_some(update.status),
// Peers replicate credentials, never the local parent binding:
// each site resolves its own parent from its own IAM.
parent_user: None,
allow_site_replicator_account,
},
)
.await
.map_err(ApiError::from)?;
return Ok(());
}
if let Some(delete) = change.delete {
if let Some(local) = iam_sys.get_user(&delete.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
iam_sys
.delete_service_account(&delete.access_key, true)
.await
.map_err(ApiError::from)?;
return Ok(());
}
Err(s3_error!(InvalidRequest, "serviceAccountChange is empty"))
}
fn claims_unix_timestamp(value: &Value) -> Option<i64> {
match value {
Value::Number(number) => number.as_i64(),
rustfs/src/admin/router.rs
+262 -8
View File
@@ -17,7 +17,8 @@ use super::storage_api::bucket::metadata_sys;
use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatus, BucketStats, ReplicationStatusType};
use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets};
use super::storage_api::bucket::target_sys::{
BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient, append_version_id_query,
BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, SsecPassthroughCapability, TargetClient,
append_version_id_query,
};
use super::storage_api::bucket::versioning_sys::BucketVersioningSys;
use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
@@ -70,6 +71,9 @@ use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy};
use rustfs_utils::http::headers::{
AMZ_CHECKSUM_CRC32, AMZ_CHECKSUM_CRC32C, AMZ_CHECKSUM_CRC64NVME, AMZ_CHECKSUM_SHA1, AMZ_CHECKSUM_SHA256, AMZ_CHECKSUM_TYPE,
};
use rustfs_utils::http::object_encryption_keys::{
REPLICATION_SSEC_ALGORITHM_HEADER, REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_SSEC_ORIGINAL_SIZE_HEADER,
};
use rustfs_utils::http::{
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header,
@@ -213,6 +217,13 @@ const REPLICATION_CHECK_ERROR_MAX_BYTES: usize = 512;
/// RustFS extension code (no madmin analogue): the target does not adopt the
/// source version id, breaking the version-identity replication contract.
const REPLICATION_CHECK_CODE_VERSION_MISMATCH: &str = "BucketRemoteTargetVersionMismatch";
/// RustFS extension code (no madmin analogue): the target drops the
/// `X-Rustfs-Replication-*` SSE-C passthrough headers, so an SSE-C replica
/// would lose its decryption material (N2 fail-closed).
const REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH: &str = "BucketRemoteSsecPassthroughUnsupported";
/// Syntactically valid stand-in SSE-C key MD5 for the passthrough probe (the
/// probe object is never decrypted; it only has to round-trip the metadata).
const REPLICATION_CHECK_SSEC_PROBE_KEY_MD5: &str = "AAAAAAAAAAAAAAAAAAAAAA==";
#[derive(Debug, Clone, serde::Serialize)]
struct ReplicationCheckResponse {
@@ -254,6 +265,8 @@ struct ReplicationCheckPhases {
put: ReplicationCheckPhaseStatus,
#[serde(rename = "VersionFidelity")]
version_fidelity: ReplicationCheckPhaseStatus,
#[serde(rename = "SsecPassthrough")]
ssec_passthrough: ReplicationCheckPhaseStatus,
#[serde(rename = "DeleteMarker")]
delete_marker: ReplicationCheckPhaseStatus,
#[serde(rename = "VersionDelete")]
@@ -1852,7 +1865,7 @@ fn build_replication_check_response(mut targets: Vec<ReplicationCheckTargetStatu
let data = serde_json::to_vec(&ReplicationCheckResponse {
status: status.to_string(),
active_mutation: true,
mutation_description: "Writes a probe object, creates a delete marker, deletes the probe version, and cleans up all probe artifacts on each target.",
mutation_description: "Writes probe objects (including an SSE-C passthrough probe), creates a delete marker, deletes the probe versions, and cleans up all probe artifacts on each target.",
probe_namespace: REPLICATION_CHECK_PROBE_PREFIX,
targets,
})
@@ -2069,6 +2082,25 @@ async fn check_replication_target(
time: OffsetDateTime::now_utc(),
};
execute_replication_probe(&mut result, &mut operations).await;
// Sync the probe verdict into the runtime capability cache: the
// replication worker then fails SSE-C replication closed on a flagged
// target (or skips its own HEAD-back audit on a proven one) without
// re-learning what the probe just established.
match (result.phases.ssec_passthrough.status, result.phases.ssec_passthrough.code) {
("OK", _) => {
BucketTargetSys::get()
.record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Supported)
.await;
}
("FAILED", Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH)) => {
BucketTargetSys::get()
.record_ssec_passthrough_capability(&target.arn, SsecPassthroughCapability::Unsupported)
.await;
}
_ => {}
}
result
}
@@ -2087,6 +2119,15 @@ struct ReplicationProbePutOutcome {
response_version_id: Option<String>,
}
/// Outcome of the SSE-C passthrough probe: whether the HEAD-back of the probe
/// replica echoed SSE-C evidence (the customer-algorithm header a RustFS
/// target restores from the passthrough transport headers), plus the version
/// the target assigned so cleanup can address it.
struct ReplicationSsecProbeOutcome {
evidence_present: bool,
version_id: Option<String>,
}
struct ReplicationProbeMultipartError {
primary: S3ClientError,
cleanup_error: Option<String>,
@@ -2109,9 +2150,14 @@ trait ReplicationProbeOperations {
/// there: a target can adopt PutObject version ids and still mint its own
/// for CreateMultipartUpload.
async fn multipart_put(&mut self) -> Result<ReplicationProbePutOutcome, ReplicationProbeMultipartError>;
/// PUT a probe version carrying the SSE-C passthrough transport headers,
/// HEAD it back through the replication-check channel, and report whether
/// the SSE-C evidence survived. Cleanup of the created version is the
/// caller's job (the outcome carries its version id).
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError>;
async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result<Option<String>, S3ClientError>;
async fn delete_version(&mut self, version_id: Option<&str>) -> Result<(), S3ClientError>;
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String>;
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String>;
}
struct RemoteReplicationProbeOperations<'a> {
@@ -2131,6 +2177,10 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> {
multipart_put_replication_probe_object(self.client, self.bucket, self.key, self.time).await
}
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
ssec_passthrough_probe_object(self.client, self.bucket, self.key, self.time).await
}
async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
delete_replication_probe_object(
self.client,
@@ -2154,7 +2204,7 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> {
.map(|_| ())
}
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> {
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> {
cleanup_replication_probe(self.client, self.bucket, self.key, known_version_ids).await
}
}
@@ -2175,6 +2225,7 @@ fn version_fidelity_error(api: &str, outcome: &ReplicationProbePutOutcome) -> Op
async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, operations: &mut impl ReplicationProbeOperations) {
let mut probe_version_id = None;
let mut multipart_probe_version_id = None;
let mut ssec_probe_version_id = None;
let mut delete_marker_version_id = None;
let mut cleanup_required = true;
let mut multipart_cleanup_error = None;
@@ -2230,6 +2281,38 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
}
}
// N2: probe SSE-C passthrough with the same transport headers live
// replication sends. A target that drops them (MinIO, generic S3) stores
// the probe as a plain object and echoes no SSE-C evidence on the
// HEAD-back; SSE-C replicas there would silently lose their decryption
// material, so the target must be flagged with a machine-readable code.
// Deliberately unlike VersionFidelity, a failed SsecPassthrough phase
// does NOT fail the target overall: version-identity drift breaks the
// replication contract for every object, while dropped SSE-C passthrough
// headers only limit a capability — a plaintext-only deployment against a
// MinIO target is perfectly healthy and must not turn red. The phase's
// own FAILED + machine-readable Code remains for madmin consumers (and
// the verdict still reaches the runtime capability cache).
if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" {
match operations.ssec_passthrough_probe().await {
Ok(outcome) => {
ssec_probe_version_id = outcome.version_id;
if outcome.evidence_present {
result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::passed();
} else {
let error = "target drops SSE-C passthrough replication headers; \
SSE-C replicas would lose their decryption material on this target";
result.phases.ssec_passthrough =
ReplicationCheckPhaseStatus::failed_with_code(error, REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH);
}
}
Err(err) => {
let error = format_replication_check_client_error(&err, ReplicationCheckFailureContext::ReplicateObject);
result.phases.ssec_passthrough = ReplicationCheckPhaseStatus::failed(&error);
}
}
}
if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" {
match operations.create_delete_marker(probe_version_id.as_deref()).await {
Ok(version_id) => {
@@ -2258,6 +2341,7 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op
.cleanup([
probe_version_id.as_deref(),
multipart_probe_version_id.as_deref(),
ssec_probe_version_id.as_deref(),
delete_marker_version_id.as_deref(),
])
.await
@@ -2552,6 +2636,72 @@ async fn put_replication_probe_object(
})
}
/// PUT a fresh probe version carrying the SSE-C passthrough transport headers
/// (the wire shape live SSE-C replication uses), then HEAD it back through the
/// worker channel (replication-check exemption + proxy suppression). A RustFS
/// target restores the transport headers into stored SSE-C metadata and its
/// HEAD echoes `x-amz-server-side-encryption-customer-algorithm`; a target
/// that dropped the headers echoes nothing. The probe body is never SSE-C
/// encrypted — only the metadata round-trip matters — and the version is
/// deleted by the shared probe cleanup.
async fn ssec_passthrough_probe_object(
target_client: &TargetClient,
target_bucket: &str,
probe_key: &str,
now: OffsetDateTime,
) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
let options = build_replication_probe_put_options(now);
let sent_version_id = options.internal.source_version_id.clone();
let mut headers = build_replication_probe_headers(&options);
// These are full wire names (not x-rustfs/x-minio suffixes), so they must
// be inserted verbatim — `insert_header` would mangle them.
for (name, value) in [
(REPLICATION_SSEC_ALGORITHM_HEADER, "AES256"),
(REPLICATION_SSEC_KEY_MD5_HEADER, REPLICATION_CHECK_SSEC_PROBE_KEY_MD5),
(REPLICATION_SSEC_ORIGINAL_SIZE_HEADER, "8"),
] {
let name = name
.parse::<HeaderName>()
.map_err(|err| S3ClientError::new(format!("invalid ssec probe header name: {err}")))?;
let value =
HeaderValue::from_str(value).map_err(|err| S3ClientError::new(format!("invalid ssec probe header value: {err}")))?;
headers.insert(name, value);
}
let query_version_id = sent_version_id.clone();
let response = target_client
.client
.put_object()
.bucket(target_bucket)
.key(probe_key)
.content_length(8)
.body(AwsByteStream::from_static(b"aaaaaaaa"))
.customize()
.map_request(move |mut req| {
for (key, value) in headers.clone() {
req.headers_mut().insert(key.expect("operation should succeed"), value);
}
let uri = append_version_id_query(req.uri(), &query_version_id);
req.set_uri(uri).map_err(std::io::Error::other)?;
Result::<_, std::io::Error>::Ok(req)
})
.send()
.await
.map_err(S3ClientError::from)?;
let version_id = response.version_id().map(ToOwned::to_owned);
let head_version = version_id.clone().or_else(|| Some(sent_version_id.clone()));
let head = target_client
.head_object(target_bucket, probe_key, head_version)
.await
.map_err(S3ClientError::from)?;
Ok(ReplicationSsecProbeOutcome {
evidence_present: head.sse_customer_algorithm().is_some_and(|algorithm| !algorithm.is_empty()),
version_id,
})
}
async fn delete_replication_probe_object(
target_client: &TargetClient,
target_bucket: &str,
@@ -3722,6 +3872,12 @@ mod tests {
/// Same, for the multipart leg: a target may mirror PutObject ids and
/// still mint its own at CreateMultipartUpload.
minted_multipart_version_id: Option<&'static str>,
/// Transport failure of the SSE-C passthrough probe itself.
ssec_probe_error: Option<&'static str>,
/// Models a MinIO-like target that drops the SSE-C passthrough
/// headers: the probe HEAD-back echoes no SSE-C evidence. The default
/// (false) models a RustFS target that preserves them.
ssec_evidence_missing: bool,
delete_marker_error: Option<&'static str>,
version_delete_error: Option<&'static str>,
cleanup_error: Option<&'static str>,
@@ -3759,6 +3915,17 @@ mod tests {
})
}
async fn ssec_passthrough_probe(&mut self) -> Result<ReplicationSsecProbeOutcome, S3ClientError> {
self.calls.push("ssec-probe");
match self.ssec_probe_error {
Some(code) => Err(scripted_probe_error(code)),
None => Ok(ReplicationSsecProbeOutcome {
evidence_present: !self.ssec_evidence_missing,
version_id: Some("ssec-version".to_string()),
}),
}
}
async fn create_delete_marker(&mut self, _version_id: Option<&str>) -> Result<Option<String>, S3ClientError> {
self.calls.push("delete-marker");
match self.delete_marker_error {
@@ -3775,7 +3942,7 @@ mod tests {
}
}
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> {
async fn cleanup(&mut self, known_version_ids: [Option<&str>; 4]) -> Result<(), String> {
self.calls.push("cleanup");
self.cleanup_ids = known_version_ids
.into_iter()
@@ -3810,8 +3977,9 @@ mod tests {
assert_eq!(result.phases.version_fidelity.code, Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH));
assert_eq!(result.phases.delete_marker.status, "SKIPPED");
assert_eq!(result.phases.version_delete.status, "SKIPPED");
assert_eq!(result.phases.ssec_passthrough.status, "SKIPPED");
assert_eq!(result.phases.cleanup.status, "OK");
assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None]);
assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None, None]);
}
#[tokio::test]
@@ -3824,6 +3992,70 @@ mod tests {
assert_eq!(result.status, "OK");
assert_eq!(result.phases.version_fidelity.status, "OK");
assert_eq!(result.phases.version_fidelity.code, None);
assert_eq!(result.phases.ssec_passthrough.status, "OK");
assert_eq!(result.phases.ssec_passthrough.code, None);
}
/// N2: a target that drops the SSE-C passthrough transport headers must
/// fail the SsecPassthrough phase with the machine-readable code while the
/// target overall stays OK — deliberately unlike VersionFidelity: this is
/// a capability limit, not a broken replication contract, and a
/// plaintext-only deployment against such a target must not turn red. The
/// other mutation phases keep running and the probe version is cleaned up.
#[tokio::test]
async fn replication_probe_flags_ssec_passthrough_dropping_target_without_failing_target() {
let mut result = replication_check_target("arn:a", "OK", None);
let mut operations = ScriptedReplicationProbe {
ssec_evidence_missing: true,
..Default::default()
};
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(result.status, "OK", "a capability-only failure must not fail the target overall");
assert_eq!(result.error, None);
assert_eq!(result.phases.ssec_passthrough.status, "FAILED");
assert_eq!(result.phases.ssec_passthrough.code, Some(REPLICATION_CHECK_CODE_SSEC_PASSTHROUGH));
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
Some("marker-version".to_string())
]
);
}
/// A transport failure of the SSE-C probe is not evidence of a dropping
/// target: the phase fails without the capability code (the runtime cache
/// stays Unknown and the worker keeps auditing), and the target overall
/// stays OK.
#[tokio::test]
async fn replication_probe_ssec_transport_failure_carries_no_capability_code() {
let mut result = replication_check_target("arn:a", "OK", None);
let mut operations = ScriptedReplicationProbe {
ssec_probe_error: Some("InternalError"),
..Default::default()
};
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(result.status, "OK");
assert_eq!(result.phases.ssec_passthrough.status, "FAILED");
assert_eq!(result.phases.ssec_passthrough.code, None);
assert_eq!(operations.cleanup_ids[2], None, "a failed ssec probe leaves no version to clean");
}
#[tokio::test]
@@ -3854,12 +4086,23 @@ mod tests {
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]);
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
None
]
);
@@ -3879,12 +4122,23 @@ mod tests {
execute_replication_probe(&mut result, &mut operations).await;
assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]);
assert_eq!(
operations.calls,
[
"put",
"multipart-put",
"ssec-probe",
"delete-marker",
"version-delete",
"cleanup"
]
);
assert_eq!(
operations.cleanup_ids,
[
Some("object-version".to_string()),
Some("multipart-version".to_string()),
Some("ssec-version".to_string()),
Some("marker-version".to_string())
]
);
rustfs/src/admin/storage_api.rs
+1
View File
@@ -196,6 +196,7 @@ pub(crate) mod bucket_target_sys {
pub(crate) type PutObjectOptions = super::ecstore_bucket::bucket_target_sys::PutObjectOptions;
pub(crate) type RemoveObjectOptions = super::ecstore_bucket::bucket_target_sys::RemoveObjectOptions;
pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError;
pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability;
pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient;
}
rustfs/src/app/object_usecase.rs
+248 -3
View File
@@ -46,9 +46,10 @@ use super::storage_api::object_usecase::bucket::{
replication::{
DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent,
delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
force_delete_target_set, has_active_delete_rule, load_delete_config_snapshot, must_replicate_object,
persist_force_delete_intent, schedule_object_replication, schedule_replication_delete, schedule_replication_deletes,
set_deleted_object_replication_state, should_schedule_delete_replication, should_use_existing_delete_replication_info,
force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, load_delete_config_snapshot,
must_replicate_object, persist_force_delete_intent, record_replication_proxy, schedule_object_replication,
schedule_replication_delete, schedule_replication_deletes, set_deleted_object_replication_state,
should_schedule_delete_replication, should_use_existing_delete_replication_info,
},
tagging::decode_tags,
validate_restore_request,
@@ -6662,6 +6663,226 @@ impl DefaultObjectUsecase {
})
}
/// Headers a proxied read forwards verbatim to the replication target:
/// only the client's SSE-C key family, so the target performs the real
/// SSE-C decryption (never the replication-check exemption). HTTP
/// conditional headers (If-Match & co.) are deliberately NOT forwarded —
/// MinIO does not forward them either, and a remote 304/412 would leak a
/// conditional evaluation against a replica the local site never saw.
/// Range and part-number travel as typed SDK parameters instead.
fn proxy_read_passthrough_headers(headers: &HeaderMap) -> HeaderMap {
const FORWARDED: &[&str] = &[
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-md5",
];
let mut forwarded = HeaderMap::new();
for name in FORWARDED {
if let Ok(header_name) = http::HeaderName::from_str(name)
&& let Some(value) = headers.get(&header_name)
{
forwarded.insert(header_name, value.clone());
}
}
forwarded
}
/// True when a proxied SDK call failed because the target does not have
/// the object either (service-level not-found or a raw 404, which also
/// covers NoSuchVersion): the caller tries the next target silently.
fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
}
/// Serve a GET whose local read failed with not-found by proxying to the
/// bucket's replication targets (MinIO `proxyGetToReplicationTarget`,
/// backlog#1675 P1-5). Returns None when no target can serve the object;
/// the caller then returns the original local error.
async fn proxy_get_object_to_replication_targets(
req: &S3Request<GetObjectInput>,
bucket: &str,
key: &str,
opts: &ObjectOptions,
) -> Option<GetObjectOutput> {
let targets = get_read_proxy_targets(bucket, key, opts).await;
if targets.is_empty() {
return None;
}
let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
let range = req
.headers
.get(http::header::RANGE)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let part_number = req.input.part_number;
for target in targets {
match target
.get_object(
&target.bucket,
key,
opts.version_id.clone(),
range.clone(),
part_number,
extra_headers.clone(),
)
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt
// (targets were available), one failed when no target
// served it — never per target.
record_replication_proxy(bucket, "GetObject", false).await;
return Some(Self::proxy_sdk_get_output_to_s3s(remote));
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: GET against replication target failed");
}
}
}
record_replication_proxy(bucket, "GetObject", true).await;
None
}
/// Serve a HEAD whose local lookup failed with not-found by proxying to
/// the bucket's replication targets (MinIO `proxyHeadToRepTarget`).
async fn proxy_head_object_to_replication_targets(
req: &S3Request<HeadObjectInput>,
bucket: &str,
key: &str,
opts: &ObjectOptions,
) -> Option<HeadObjectOutput> {
let targets = get_read_proxy_targets(bucket, key, opts).await;
if targets.is_empty() {
return None;
}
let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
let range = req
.headers
.get(http::header::RANGE)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let part_number = req.input.part_number;
for target in targets {
match target
.head_object_for_proxy(
&target.bucket,
key,
opts.version_id.clone(),
range.clone(),
part_number,
extra_headers.clone(),
)
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "HeadObject", false).await;
return Some(Self::proxy_sdk_head_output_to_s3s(remote));
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: HEAD against replication target failed");
}
}
}
record_replication_proxy(bucket, "HeadObject", true).await;
None
}
/// Translate a proxied SDK GET response into the s3s output, forwarding
/// the body as a stream (no buffering, no local persistence).
fn proxy_sdk_get_output_to_s3s(remote: aws_sdk_s3::operation::get_object::GetObjectOutput) -> GetObjectOutput {
let body = remote.body;
let body_stream = tokio_util::io::ReaderStream::with_capacity(body.into_async_read(), 64 * 1024);
GetObjectOutput {
body: Some(StreamingBlob::wrap(body_stream)),
content_length: remote.content_length,
content_range: remote.content_range,
content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
content_encoding: remote.content_encoding,
content_disposition: remote.content_disposition,
content_language: remote.content_language,
cache_control: remote.cache_control,
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
last_modified: remote
.last_modified
.and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
.map(Timestamp::from),
metadata: remote.metadata,
version_id: remote.version_id,
server_side_encryption: remote
.server_side_encryption
.map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
sse_customer_algorithm: remote.sse_customer_algorithm,
sse_customer_key_md5: remote.sse_customer_key_md5,
ssekms_key_id: remote.ssekms_key_id,
parts_count: remote.parts_count,
tag_count: remote.tag_count,
storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
expiration: remote.expiration,
restore: remote.restore,
checksum_crc32: remote.checksum_crc32,
checksum_crc32c: remote.checksum_crc32_c,
checksum_crc64nvme: remote.checksum_crc64_nvme,
checksum_sha1: remote.checksum_sha1,
checksum_sha256: remote.checksum_sha256,
checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
..Default::default()
}
}
/// Translate a proxied SDK HEAD response into the s3s output.
///
/// Known gaps: the SDK's HeadObjectOutput does not model 206/Content-Range
/// for a ranged HEAD (the SDK exposes no content_range member on HEAD),
/// and s3s' typed HeadObjectOutput has no tag_count field (the local path
/// injects x-amz-tagging-count as a raw header) — both are dropped for
/// proxied HEADs.
fn proxy_sdk_head_output_to_s3s(remote: aws_sdk_s3::operation::head_object::HeadObjectOutput) -> HeadObjectOutput {
HeadObjectOutput {
content_length: remote.content_length,
content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
content_encoding: remote.content_encoding,
content_disposition: remote.content_disposition,
content_language: remote.content_language,
cache_control: remote.cache_control,
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
last_modified: remote
.last_modified
.and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
.map(Timestamp::from),
metadata: remote.metadata,
version_id: remote.version_id,
server_side_encryption: remote
.server_side_encryption
.map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
sse_customer_algorithm: remote.sse_customer_algorithm,
sse_customer_key_md5: remote.sse_customer_key_md5,
ssekms_key_id: remote.ssekms_key_id,
parts_count: remote.parts_count,
storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
expiration: remote.expiration,
restore: remote.restore,
checksum_crc32: remote.checksum_crc32,
checksum_crc32c: remote.checksum_crc32_c,
checksum_crc64nvme: remote.checksum_crc64_nvme,
checksum_sha1: remote.checksum_sha1,
checksum_sha256: remote.checksum_sha256,
checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
..Default::default()
}
}
#[instrument(name = "execute_get_object", level = "trace", skip(self, req))]
pub async fn execute_get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
self.execute_get_object_boxed(req).await
@@ -6787,6 +7008,19 @@ impl DefaultObjectUsecase {
{
Ok(prepared_read) => prepared_read,
Err(err) => {
// Active-active replication lag window: an object missing
// locally (and only missing — other errors keep their
// semantics) may still be served by proxying the GET to a
// replication target (backlog#1675 P1-5).
if matches!(*err.code(), S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion)
&& let Some(output) = Self::proxy_get_object_to_replication_targets(&req, &bucket, &key, &opts).await
{
lifecycle.finish_ok();
let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
let result = Ok(response);
let _ = helper.version_id(version_id_for_event).complete(&result);
return result;
}
lifecycle.finish_err();
return Err(err);
}
@@ -8696,6 +8930,17 @@ impl DefaultObjectUsecase {
let msg = head_prefix_not_found_message(&bucket, &key, has_children);
return Err(S3Error::with_message(S3ErrorCode::NoSuchKey, msg));
}
// Active-active replication lag window: an object missing
// locally may still be served by proxying the HEAD to a
// replication target (backlog#1675 P1-5).
if let Some(output) = Self::proxy_head_object_to_replication_targets(&req, &bucket, &key, &opts).await {
let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
let result = Ok(response);
let _ = helper
.version_id(req.input.version_id.clone().unwrap_or_default())
.complete(&result);
return result;
}
return Err(S3Error::new(S3ErrorCode::NoSuchKey));
}
// Other errors, such as insufficient permissions, still return the original error
rustfs/src/app/storage_api.rs
+18
View File
@@ -627,6 +627,24 @@ pub(crate) mod bucket {
#[cfg(test)]
pub(crate) use replication_contracts::replication_statuses_map;
/// Remote replication-target client used by the read-proxy path.
pub(crate) type ProxyTargetClient = crate::storage::storage_api::ecstore_bucket::bucket_target_sys::TargetClient;
/// Proxy-request metric recorder (get/head/tagging totals + failures).
pub(crate) use crate::storage::storage_api::record_replication_proxy;
/// Replication targets eligible to serve a proxied GET/HEAD/Tagging of
/// an object not present locally (MinIO `getProxyTargets`; empty when
/// the request was itself proxied, versioning is suspended, or no
/// replication rule matches). backlog#1675 P1-5.
pub(crate) async fn get_read_proxy_targets(
bucket: &str,
object: &str,
opts: &crate::storage::storage_api::StorageObjectOptions,
) -> Vec<Arc<ProxyTargetClient>> {
replication_contracts::get_proxy_targets(bucket, object, opts).await
}
pub(crate) async fn persist_force_delete_intent(
store: Arc<crate::storage::storage_api::ECStore>,
bucket: String,
rustfs/src/storage/ecfs.rs
+229 -38
View File
@@ -12,15 +12,15 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::StorageVersioningConfigExt as _;
use super::{
BUCKET_ACCELERATE_CONFIG, BUCKET_LOGGING_CONFIG, BUCKET_REQUEST_PAYMENT_CONFIG, BUCKET_VERSIONING_CONFIG,
BUCKET_WEBSITE_CONFIG, BucketVersioningSys, OBJECT_LOCK_CONFIG, StorageError, check_retention_for_modification, decode_tags,
decode_tags_to_map, delete_bucket_metadata_config_if_incarnation, encode_tags, get_bucket_accelerate_config,
get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_replication_config, get_bucket_request_payment_config,
get_bucket_website_config, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
record_replication_proxy, serialize, update_bucket_metadata_config_if_incarnation,
get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_website_config,
is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, record_replication_proxy, serialize,
update_bucket_metadata_config_if_incarnation,
};
use super::{StorageReplicationConfigExt as _, StorageVersioningConfigExt as _};
use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
use crate::error::ApiError;
use crate::storage::access::{apply_bucket_generation_guard, bucket_config_mutation_incarnation, has_bypass_governance_header};
@@ -59,7 +59,7 @@ const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock";
const LOG_SUBSYSTEM_TAGGING: &str = "tagging";
use crate::app::storage_api::object_usecase::bucket::replication::{
ReplicateDecision, must_replicate_metadata, schedule_metadata_replication,
ReplicateDecision, get_read_proxy_targets, must_replicate_metadata, schedule_metadata_replication,
};
use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;
@@ -105,18 +105,152 @@ impl FS {
&self.server_ctx
}
async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool {
get_bucket_replication_config(bucket)
.await
.map(|(cfg, _)| cfg.has_active_rules(object, true))
.unwrap_or(false)
/// Not-found classifier for proxied SDK tagging calls: a raw 404 covers
/// NoSuchKey and NoSuchVersion alike; the caller silently tries the next
/// replication target.
fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
}
async fn record_replication_tagging_metric(bucket: &str, object: &str, api: &str, is_err: bool) {
if !Self::replication_tagging_enabled(bucket, object).await {
return;
/// Selector options for a tagging proxy. Reuses `get_opts` so the
/// anti-loop `source-proxy-request` header family and the bucket's
/// version-suspension state gate proxying exactly like GET/HEAD.
async fn tagging_proxy_opts(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<ObjectOptions> {
get_opts(bucket, object, version_id, None, headers).await.ok()
}
/// Serve a GetObjectTagging for an object missing locally by proxying to
/// the bucket's replication targets (MinIO `proxyGetTaggingToRepTarget`,
/// backlog#1675 P1-5). None means no target had the object.
async fn proxy_get_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<TagSet> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
}
record_replication_proxy(bucket, api, is_err).await;
for target in targets {
match target
.get_object_tagging(&target.bucket, object, opts.version_id.clone())
.await
{
Ok(remote) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "GetObjectTagging", false).await;
return Some(
remote
.tag_set
.into_iter()
.map(|tag| Tag {
key: Some(tag.key),
value: Some(tag.value),
})
.collect(),
);
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: GetObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "GetObjectTagging", true).await;
None
}
/// Apply a PutObjectTagging for an object missing locally on a
/// replication target (MinIO `proxyTaggingToRepTarget`).
async fn proxy_put_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
tag_set: &TagSet,
) -> Option<()> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let mut tagging = aws_sdk_s3::types::Tagging::builder();
for tag in tag_set {
let sdk_tag = aws_sdk_s3::types::Tag::builder()
.key(tag.key.clone().unwrap_or_default())
.value(tag.value.clone().unwrap_or_default())
.build()
.ok()?;
tagging = tagging.tag_set(sdk_tag);
}
let tagging = tagging.build().ok()?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
}
for target in targets {
match target
.put_object_tagging(&target.bucket, object, opts.version_id.clone(), tagging.clone())
.await
{
Ok(_) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "PutObjectTagging", false).await;
return Some(());
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: PutObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "PutObjectTagging", true).await;
None
}
/// Apply a DeleteObjectTagging for an object missing locally on a
/// replication target (MinIO `proxyTaggingToRepTarget`).
async fn proxy_delete_object_tagging(
bucket: &str,
object: &str,
version_id: Option<String>,
headers: &http::HeaderMap,
) -> Option<()> {
let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
let targets = get_read_proxy_targets(bucket, object, &opts).await;
if targets.is_empty() {
return None;
}
for target in targets {
match target
.delete_object_tagging(&target.bucket, object, opts.version_id.clone())
.await
{
Ok(_) => {
// MinIO-aligned accounting: one total per proxy attempt,
// one failed when no target served it.
record_replication_proxy(bucket, "DeleteObjectTagging", false).await;
return Some(());
}
Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
}
Err(err) => {
warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: DeleteObjectTagging against replication target failed");
}
}
}
record_replication_proxy(bucket, "DeleteObjectTagging", true).await;
None
}
pub async fn get_object_tag_conditions_for_policy(
@@ -447,7 +581,27 @@ impl S3 for FS {
let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?;
let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
Ok(info) => info,
Err(e) => {
// Replication lag window: apply the tagging delete on a
// replication target that already has the object
// (backlog#1675 P1-5). No local object exists, so no bucket
// notification event is emitted for the proxied write.
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& Self::proxy_delete_object_tagging(&bucket, &object, version_id.clone(), &req.headers)
.await
.is_some()
{
counter!("rustfs_delete_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete")
.record(duration.as_secs_f64());
return Ok(S3Response::new(DeleteObjectTaggingOutput { version_id }));
}
return Err(ApiError::from(e).into());
}
};
let dsc = must_replicate_metadata(
&bucket,
&object,
@@ -470,7 +624,6 @@ impl S3 for FS {
}
let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await;
Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await;
let object_info = delete_tags_result.map_err(|e| {
error!(
component = LOG_COMPONENT_STORAGE,
@@ -928,32 +1081,49 @@ impl S3 for FS {
..Default::default()
};
let tags_result = store.get_object_tags(bucket, object, &opts).await;
Self::record_replication_tagging_metric(bucket, object, "GetObjectTagging", tags_result.is_err()).await;
let tags = tags_result.map_err(|e| {
if is_err_object_not_found(&e) {
debug!(
let tags = match store.get_object_tags(bucket, object, &opts).await {
Ok(tags) => tags,
Err(e) => {
// Replication lag window: the object may exist on a
// replication target even though it is missing locally —
// proxy the tagging read there (backlog#1675 P1-5).
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& let Some(tag_set) =
Self::proxy_get_object_tagging(bucket, object, req.input.version_id.clone(), &req.headers).await
{
counter!("rustfs_get_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "get")
.record(duration.as_secs_f64());
return Ok(S3Response::new(GetObjectTaggingOutput {
tag_set,
version_id: req.input.version_id.clone(),
}));
}
if is_err_object_not_found(&e) {
debug!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_not_found",
bucket = %bucket,
object = %object,
error = %e,
"Object tags not found"
);
return Err(s3_error!(NoSuchKey));
}
error!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_not_found",
event = "object_tagging_get_failed",
bucket = %bucket,
object = %object,
error = %e,
"Object tags not found"
"Failed to load object tags"
);
return s3_error!(NoSuchKey);
return Err(ApiError::from(e).into());
}
error!(
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_TAGGING,
event = "object_tagging_get_failed",
bucket = %bucket,
object = %object,
error = %e,
"Failed to load object tags"
);
ApiError::from(e).into()
})?;
};
let tag_set = decode_tags(tags.as_str());
debug!(
@@ -1629,14 +1799,36 @@ impl S3 for FS {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let tags = encode_tags(tagging.tag_set);
let tags = encode_tags(tagging.tag_set.clone());
debug!("Encoded tags: {}", tags);
let version_id = req.input.version_id.clone();
let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?;
let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
Ok(info) => info,
Err(e) => {
// Replication lag window: apply the tagging update on a
// replication target that already has the object
// (backlog#1675 P1-5). No local object exists, so no bucket
// notification event is emitted for the proxied write.
if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
&& Self::proxy_put_object_tagging(&bucket, &object, version_id.clone(), &req.headers, &tagging.tag_set)
.await
.is_some()
{
counter!("rustfs_put_object_tagging_success").increment(1);
let duration = start_time.elapsed();
histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put")
.record(duration.as_secs_f64());
return Ok(S3Response::new(PutObjectTaggingOutput {
version_id: req.input.version_id.clone(),
}));
}
return Err(ApiError::from(e).into());
}
};
let dsc = must_replicate_metadata(
&bucket,
&object,
@@ -1659,7 +1851,6 @@ impl S3 for FS {
}
let put_tags_result = store.put_object_tags(&bucket, &object, &tags, &opts).await;
Self::record_replication_tagging_metric(&bucket, &object, "PutObjectTagging", put_tags_result.is_err()).await;
let object_info = put_tags_result.map_err(|e| {
error!("Failed to put object tags: {}", e);
counter!("rustfs_put_object_tagging_failure").increment(1);
rustfs/src/storage/mod.rs
+18 -18
View File
@@ -55,24 +55,24 @@ pub(crate) use storage_api::{
QuotaError, RUSTFS_META_BUCKET, RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
ReplicationStats, ReplicationStatusType, Result, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
StorageDeletedObject, StorageDiskRpcExt, StorageError, StorageGetObjectReader, StorageObjectInfo, StorageObjectOptions,
StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageReplicationConfigExt, StorageVersioningConfigExt,
TONIC_RPC_PREFIX, TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef,
WriteEncryption, WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path,
check_retention_for_modification, collect_local_metrics, compression_metadata_value, contract, decode_tags,
decode_tags_to_map, delete_bucket_metadata_config, delete_bucket_metadata_config_if_incarnation, disk_drive_path,
disk_endpoint, ecfs_consumer, ecfs_extend_consumer, ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client,
ecstore_cluster, ecstore_compression, ecstore_config, ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event,
ecstore_layout, ecstore_metrics, ecstore_notification, ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk,
ecstore_storage, ecstore_tier, encode_tags, find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config,
get_bucket_logging_config, get_bucket_metadata, get_bucket_notification_config, get_bucket_object_lock_config,
get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config,
get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer, init_background_replication,
init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class, options_consumer,
prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer, runtime_sources_consumer,
s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities,
try_migrate_bucket_metadata, try_migrate_iam_config, try_migrate_server_config, update_bucket_metadata_config,
update_bucket_metadata_config_if_incarnation, verify_rpc_signature, wrap_reader,
StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageVersioningConfigExt, TONIC_RPC_PREFIX,
TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef, WriteEncryption,
WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path, check_retention_for_modification,
collect_local_metrics, compression_metadata_value, contract, decode_tags, decode_tags_to_map, delete_bucket_metadata_config,
delete_bucket_metadata_config_if_incarnation, disk_drive_path, disk_endpoint, ecfs_consumer, ecfs_extend_consumer,
ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client, ecstore_cluster, ecstore_compression, ecstore_config,
ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event, ecstore_layout, ecstore_metrics, ecstore_notification,
ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk, ecstore_storage, ecstore_tier, encode_tags,
find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config, get_bucket_logging_config, get_bucket_metadata,
get_bucket_notification_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_sse_config,
get_bucket_website_config, get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer,
init_background_replication, init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx,
init_lock_clients, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class,
options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer,
runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag,
topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
try_migrate_server_config, update_bucket_metadata_config, update_bucket_metadata_config_if_incarnation, verify_rpc_signature,
wrap_reader,
};
#[cfg(test)]
rustfs/src/storage/options.rs
+93 -3
View File
@@ -19,9 +19,10 @@ use http::{HeaderMap, HeaderValue};
use rustfs_utils::http::{
AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
SUFFIX_TAGGING_TIMESTAMP, get_header,
header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
insert_header_map, insert_str,
metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
@@ -276,6 +277,19 @@ pub async fn get_opts(
// Background scanner still performs full integrity checks asynchronously.
opts.skip_verify_bitrot = get_skip_verify_bitrot();
// Anti-loop markers for the replication read proxy
// (`{x-rustfs-,x-minio-}source-proxy-request` header family).
// MinIO semantics: the header being PRESENT at all (`ProxyHeaderSet`)
// disables proxying, whatever its value — a peer's replication worker
// sends "false" on its convergence HEADs so the receiver answers locally
// instead of proxying the miss back (a proxied echo would fake
// convergence and the object would never replicate). Deliberately not
// gated on replication authorization: the header only disables proxying
// (it grants nothing).
let proxy_header = get_header(headers, SUFFIX_SOURCE_PROXY_REQUEST);
opts.proxy_header_set = proxy_header.is_some();
opts.proxy_request = proxy_header.map(|v| v.as_ref() == "true").unwrap_or_default();
fill_conditional_writes_opts_from_header(headers, &mut opts)?;
Ok(opts)
@@ -2544,4 +2558,80 @@ mod tests {
}
}
}
/// The replication read-proxy anti-loop markers must be honored under
/// both interop prefixes (a MinIO peer sends x-minio-, a RustFS peer
/// sends both). `proxy_request` is set only for the literal value
/// "true", while `proxy_header_set` (MinIO `ProxyHeaderSet`) is set by
/// the header's mere presence — "false" (the replication worker's
/// convergence-HEAD marker) and arbitrary values included — so the
/// selector refuses to proxy either way.
#[tokio::test]
async fn test_get_opts_parses_source_proxy_request_under_both_prefixes() {
for header_name in ["x-rustfs-source-proxy-request", "x-minio-source-proxy-request"] {
let mut headers = HeaderMap::new();
headers.insert(header_name, HeaderValue::from_static("true"));
let opts = get_opts("test-bucket", "test-object", None, None, &headers)
.await
.expect("get_opts should succeed");
assert!(opts.proxy_request, "{header_name} must set opts.proxy_request");
assert!(opts.proxy_header_set, "{header_name} must set opts.proxy_header_set");
}
let opts = get_opts("test-bucket", "test-object", None, None, &HeaderMap::new())
.await
.expect("get_opts should succeed");
assert!(!opts.proxy_request, "absent header must leave proxy_request off");
assert!(!opts.proxy_header_set, "absent header must leave proxy_header_set off");
for (header_name, value) in [
("x-minio-source-proxy-request", "false"),
("x-rustfs-source-proxy-request", "false"),
("x-minio-source-proxy-request", "anything-else"),
] {
let mut headers = HeaderMap::new();
headers.insert(header_name, HeaderValue::from_static(value));
let opts = get_opts("test-bucket", "test-object", None, None, &headers)
.await
.expect("get_opts should succeed");
assert!(!opts.proxy_request, "{header_name}: non-'true' value must leave proxy_request off");
assert!(
opts.proxy_header_set,
"{header_name}: value {value:?} must still set proxy_header_set (presence disables proxying)"
);
}
}
/// Pin that the source-proxy-request transport family cannot be
/// materialized as bare stored metadata via an `x-*-meta-` disguise: the
/// reserved-key namespacing (`x-rustfs-source-` / `x-minio-source-`
/// prefixes in `is_reserved_user_metadata_key`) must keep covering it.
#[test]
fn test_source_proxy_request_family_is_reserved_user_metadata() {
let mut headers = HeaderMap::new();
headers.insert("x-amz-meta-x-minio-source-proxy-request", HeaderValue::from_static("true"));
headers.insert("x-rustfs-meta-x-rustfs-source-proxy-request", HeaderValue::from_static("true"));
// The bare transport header itself is not a user-metadata prefix and
// must never land in stored metadata at all.
headers.insert("x-minio-source-proxy-request", HeaderValue::from_static("true"));
let metadata = extract_metadata(&headers);
assert!(
!metadata.contains_key("x-minio-source-proxy-request"),
"bare source-proxy-request key must not be storable: {metadata:?}"
);
assert!(
!metadata.contains_key("x-rustfs-source-proxy-request"),
"bare source-proxy-request key must not be storable: {metadata:?}"
);
assert!(
metadata.contains_key("x-amz-meta-x-minio-source-proxy-request"),
"disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
);
assert!(
metadata.contains_key("x-amz-meta-x-rustfs-source-proxy-request"),
"disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
);
}
}
rustfs/src/storage/storage_api.rs
+8 -18
View File
@@ -809,6 +809,10 @@ impl StorageReplicationStatsHandle {
proxy_head_failed: metrics.proxied.head_failed,
proxy_put_tag_total: metrics.proxied.put_tag_total,
proxy_put_tag_failed: metrics.proxied.put_tag_failed,
proxy_get_tag_total: metrics.proxied.get_tag_total,
proxy_get_tag_failed: metrics.proxied.get_tag_failed,
proxy_delete_tag_total: metrics.proxied.delete_tag_total,
proxy_delete_tag_failed: metrics.proxied.delete_tag_failed,
replica_size: metrics.replica_size,
replica_count: metrics.replica_count,
}
@@ -845,6 +849,10 @@ pub(crate) struct ReplicationSiteMetricsSnapshot {
pub(crate) proxy_head_failed: i64,
pub(crate) proxy_put_tag_total: i64,
pub(crate) proxy_put_tag_failed: i64,
pub(crate) proxy_get_tag_total: i64,
pub(crate) proxy_get_tag_failed: i64,
pub(crate) proxy_delete_tag_total: i64,
pub(crate) proxy_delete_tag_failed: i64,
pub(crate) replica_size: i64,
pub(crate) replica_count: i64,
}
@@ -1491,12 +1499,6 @@ pub(crate) async fn get_bucket_object_lock_config(
ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
}
pub(crate) async fn get_bucket_replication_config(
bucket: &str,
) -> Result<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_replication_config(bucket).await
}
pub(crate) async fn persist_force_delete_intent(
api: Arc<ECStore>,
entry: ecstore_bucket::replication::MrfReplicateEntry,
@@ -1842,18 +1844,6 @@ pub(crate) async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore>
ecstore_storage::find_local_disk_by_ref(disk_ref).await
}
pub(crate) trait StorageReplicationConfigExt {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
}
impl StorageReplicationConfigExt for s3s::dto::ReplicationConfiguration {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
<s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
self, prefix, recursive,
)
}
}
pub(crate) trait StorageVersioningConfigExt {
fn enabled(&self) -> bool;
}