gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 13:27:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(site-replication): send peer join to MinIO peer/join route with encrypted payload (#5748)

Browse Source 
* test(site-replication): pin peer join wire path to MinIO peer/join route

MinIO only ever registered PUT /minio/admin/v3/site-replication/peer/join;
the /site-replication/join path never existed upstream. Flip the wire-path
and payload-encryption expectations to the real MinIO route. These tests
fail until the outbound rewrite is fixed.

* fix(site-replication): send peer join to MinIO peer/join route with encrypted payload

MinIO only registers PUT /minio/admin/v3/site-replication/peer/join; the
/site-replication/join path never existed upstream, so the outbound join
special-case rewrote requests to a 404 route. Drop the special case so
peer join falls into the generic /rustfs -> /minio prefix rewrite, and
move the payload-encryption predicate to the peer/join route (MinIO's
SRPeerJoin force-decrypts the request body).

MinIO also replies with an empty body on a successful join, which the
previous strict JSON parse rejected. Tolerate an empty/whitespace body by
synthesizing the peer identity from the add preflight metainfo
(deployment id) already fetched for the site.

Inbound dual-path registration (join and peer/join under both admin
prefixes) is intentionally unchanged for rolling upgrades from older
RustFS peers that still send the legacy outbound path.
This commit is contained in:
唐小鸭
2026-08-06 04:15:35 +08:00
committed by GitHub
parent ea8dbf49a2
commit 066e952df1
1 changed files with 54 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/site_replication.rs
+54 -10
View File
@@ -152,7 +152,7 @@ const SITE_REPLICATION_PEER_REMOVE_PATH: &str = "/rustfs/admin/v3/site-replicati
const SITE_REPLICATION_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull";
const RUSTFS_ADMIN_V3_PREFIX: &str = "/rustfs/admin/v3";
const MINIO_ADMIN_V3_PREFIX: &str = "/minio/admin/v3";
const MINIO_SITE_REPLICATION_JOIN_PATH: &str = "/minio/admin/v3/site-replication/join";
const MINIO_SITE_REPLICATION_PEER_JOIN_PATH: &str = "/minio/admin/v3/site-replication/peer/join";
fn site_replicator_service_account_policy() -> S3Result<Policy> {
Policy::parse_config(
@@ -2882,9 +2882,7 @@ fn site_replication_peer_wire_path(path: &str) -> String {
.split_once('?')
.map(|(path, query)| (path, Some(query)))
.unwrap_or((path, None));
let wire_path = if path_only == SITE_REPLICATION_PEER_JOIN_PATH {
MINIO_SITE_REPLICATION_JOIN_PATH.to_string()
} else if let Some(suffix) = path_only.strip_prefix(RUSTFS_ADMIN_V3_PREFIX) {
let wire_path = if let Some(suffix) = path_only.strip_prefix(RUSTFS_ADMIN_V3_PREFIX) {
format!("{MINIO_ADMIN_V3_PREFIX}{suffix}")
} else {
path_only.to_string()
@@ -2897,7 +2895,9 @@ fn site_replication_peer_wire_path(path: &str) -> String {
}
fn site_replication_peer_payload_encrypted(wire_path: &str) -> bool {
wire_path.split_once('?').map(|(path, _)| path).unwrap_or(wire_path) == MINIO_SITE_REPLICATION_JOIN_PATH
// MinIO's SRPeerJoin handler force-decrypts the request body, so the
// peer/join payload must always travel encrypted.
wire_path.split_once('?').map(|(path, _)| path).unwrap_or(wire_path) == MINIO_SITE_REPLICATION_PEER_JOIN_PATH
}
fn site_replication_peer_payload(path: &str, secret_key: &str, payload: Vec<u8>) -> S3Result<(Vec<u8>, &'static str)> {
@@ -8079,6 +8079,18 @@ fn sts_replication_compatibility_policy<'a>(claims: &HashMap<String, Value>, par
pub struct SiteReplicationAddHandler {}
/// MinIO's `SRPeerJoin` replies with an empty body on success; synthesize the
/// peer identity from the add preflight metainfo in that case.
fn parse_peer_join_response(body: &[u8], fallback_peer: PeerInfo) -> Result<SRPeerJoinResponse, serde_json::Error> {
if body.iter().all(u8::is_ascii_whitespace) {
return Ok(SRPeerJoinResponse {
peer: fallback_peer,
initial_sync_error_message: String::new(),
});
}
serde_json::from_slice(body)
}
#[async_trait::async_trait]
impl Operation for SiteReplicationAddHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -8147,7 +8159,7 @@ impl Operation for SiteReplicationAddHandler {
let mut joined_endpoints = HashSet::new();
let mut initial_sync_errors = SiteReplicationErrorSummary::default();
for site in &sites {
for (site, preflight) in sites.iter().zip(preflight_infos.iter()) {
if same_identity_endpoint(&site.endpoint, &local_peer.endpoint)
|| !joined_endpoints.insert(site_identity_key(&site.endpoint))
{
@@ -8160,7 +8172,10 @@ impl Operation for SiteReplicationAddHandler {
let body =
send_peer_admin_request(&connection, &peer_join_path, &site.access_key, &site.secret_key, &peer_join_req).await?;
let join_response: SRPeerJoinResponse = serde_json::from_slice(&body).map_err(|e| {
let mut fallback_peer = existing_peer_for_endpoint(&state, &site.endpoint)
.unwrap_or_else(|| normalize_peer_site(site.clone(), replicate_ilm_expiry));
fallback_peer.deployment_id = preflight.deployment_id.clone();
let join_response = parse_peer_join_response(&body, fallback_peer).map_err(|e| {
S3Error::with_message(
S3ErrorCode::InternalError,
format!("parse peer join response from {} failed: {e}", site.endpoint),
@@ -13683,7 +13698,7 @@ mod tests {
fn test_site_replication_peer_wire_path_matches_minio_routes() {
assert_eq!(
site_replication_peer_wire_path(SITE_REPLICATION_PEER_JOIN_PATH),
"/minio/admin/v3/site-replication/join"
"/minio/admin/v3/site-replication/peer/join"
);
assert_eq!(
site_replication_peer_wire_path("/rustfs/admin/v3/site-replication/peer/bucket-meta"),
@@ -13697,16 +13712,45 @@ mod tests {
#[test]
fn test_site_replication_peer_payload_encryption_matches_minio_contract() {
assert!(site_replication_peer_payload_encrypted("/minio/admin/v3/site-replication/join"));
assert!(site_replication_peer_payload_encrypted("/minio/admin/v3/site-replication/peer/join"));
assert!(site_replication_peer_payload_encrypted(
"/minio/admin/v3/site-replication/join?replicateILMExpiry=true"
"/minio/admin/v3/site-replication/peer/join?bootstrapToken=token"
));
// The outbound rewrite no longer produces the legacy `/site-replication/join`
// path; it must not be treated as an encrypted MinIO route.
assert!(!site_replication_peer_payload_encrypted("/minio/admin/v3/site-replication/join"));
assert!(!site_replication_peer_payload_encrypted(
"/minio/admin/v3/site-replication/peer/bucket-meta"
));
assert!(!site_replication_peer_payload_encrypted("/minio/admin/v3/site-replication/peer/iam-item"));
}
#[test]
fn test_parse_peer_join_response_tolerates_empty_minio_success_body() {
let fallback = PeerInfo {
deployment_id: "remote-deployment".to_string(),
..peer("remote", "https://remote.example.com")
};
for body in [&b""[..], b" \r\n\t "] {
let response = parse_peer_join_response(body, fallback.clone()).expect("empty join body is a MinIO success");
assert_eq!(response.peer.deployment_id, "remote-deployment");
assert_eq!(response.peer.endpoint, "https://remote.example.com");
assert!(response.initial_sync_error_message.is_empty());
}
let json = serde_json::to_vec(&SRPeerJoinResponse {
peer: peer("actual", "https://actual.example.com"),
initial_sync_error_message: "sync failed".to_string(),
})
.expect("serialize join response");
let response = parse_peer_join_response(&json, fallback.clone()).expect("parse join response body");
assert_eq!(response.peer.endpoint, "https://actual.example.com");
assert_eq!(response.initial_sync_error_message, "sync failed");
assert!(parse_peer_join_response(b"not-json", fallback).is_err());
}
#[test]
fn test_secret_candidate_retry_only_for_auth_errors() {
assert!(peer_error_may_be_secret_mismatch(