fix(security): redact OIDC secrets from logs and errors (#5147 )

Stop the OIDC subsystem from writing credential-grade secrets into logs and returned errors. - crates/iam/src/oidc.rs: redact sensitive header values (authorization, proxy-authorization, cookie, set-cookie) in format_http_headers, emitting only name and length; drop the raw request/response body from the DEBUG events (keep byte length); stop logging and stop splicing the raw token response body into the error returned on token_response_parse_failed (the TokenResponseBodyShape summary and length are retained); remove the now-unused format_http_body helper. - rustfs/src/admin/handlers/oidc.rs: stop logging the raw authorization code and state on the code-exchange error path (code_len/state_len are kept). This is the OIDC log/error redaction pre-work (batch 0B) from the OIDC review in rustfs/backlog#1437. It changes diagnostic content only; HTTP/STS status codes and legitimate request results are unchanged.
chore(release): prepare 1.0.0-beta.11 (#5146 )
2026-07-30 10:08:58 +00:00 · 2026-07-23 01:09:53 +00:00 · 2026-07-23 01:09:29 +00:00 · 2026-07-23 08:48:28 +08:00 · 2026-07-23 08:31:21 +08:00 · 2026-07-23 08:02:50 +08:00
36 changed files with 3816 additions and 456 deletions
@@ -3648,7 +3648,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"

 [[package]]
 name = "e2e_test"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "anyhow",
 "astral-tokio-tar",
@@ -3685,6 +3685,7 @@ dependencies = [
 "rustfs-protos",
 "rustfs-rio",
 "rustfs-signer",
+ "rustfs-utils",
 "rustls",
 "s3s",
 "serde",
@@ -8824,7 +8825,7 @@ dependencies = [

 [[package]]
 name = "rustfs"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "anyhow",
@@ -8958,7 +8959,7 @@ dependencies = [

 [[package]]
 name = "rustfs-audit"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "chrono",
@@ -8980,7 +8981,7 @@ dependencies = [

 [[package]]
 name = "rustfs-checksums"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -8995,7 +8996,7 @@ dependencies = [

 [[package]]
 name = "rustfs-common"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "chrono",
 "metrics",
@@ -9010,7 +9011,7 @@ dependencies = [

 [[package]]
 name = "rustfs-concurrency"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "insta",
 "rustfs-io-core",
@@ -9022,7 +9023,7 @@ dependencies = [

 [[package]]
 name = "rustfs-config"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "const-str",
 "serde",
@@ -9031,7 +9032,7 @@ dependencies = [

 [[package]]
 name = "rustfs-credentials"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "base64-simd",
 "hmac 0.13.0",
@@ -9044,7 +9045,7 @@ dependencies = [

 [[package]]
 name = "rustfs-crypto"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "argon2",
@@ -9064,7 +9065,7 @@ dependencies = [

 [[package]]
 name = "rustfs-data-usage"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "path-clean",
@@ -9075,7 +9076,7 @@ dependencies = [

 [[package]]
 name = "rustfs-ecstore"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "arc-swap",
@@ -9210,7 +9211,7 @@ dependencies = [

 [[package]]
 name = "rustfs-extension-schema"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "serde",
 "serde_json",
@@ -9219,7 +9220,7 @@ dependencies = [

 [[package]]
 name = "rustfs-filemeta"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "arc-swap",
 "byteorder",
@@ -9245,7 +9246,7 @@ dependencies = [

 [[package]]
 name = "rustfs-heal"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "base64 0.22.1",
@@ -9275,7 +9276,7 @@ dependencies = [

 [[package]]
 name = "rustfs-iam"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -9312,7 +9313,7 @@ dependencies = [

 [[package]]
 name = "rustfs-io-core"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "bytes",
 "memmap2",
@@ -9324,7 +9325,7 @@ dependencies = [

 [[package]]
 name = "rustfs-io-metrics"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "criterion",
 "metrics",
@@ -9387,7 +9388,7 @@ dependencies = [

 [[package]]
 name = "rustfs-keystone"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "bytes",
 "futures",
@@ -9413,7 +9414,7 @@ dependencies = [

 [[package]]
 name = "rustfs-kms"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "arc-swap",
@@ -9444,7 +9445,7 @@ dependencies = [

 [[package]]
 name = "rustfs-lifecycle"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "proptest",
@@ -9464,7 +9465,7 @@ dependencies = [

 [[package]]
 name = "rustfs-lock"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "crossbeam-queue",
@@ -9485,7 +9486,7 @@ dependencies = [

 [[package]]
 name = "rustfs-log-analyzer"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "chrono",
 "flate2",
@@ -9503,7 +9504,7 @@ dependencies = [

 [[package]]
 name = "rustfs-madmin"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "chrono",
 "humantime",
@@ -9517,7 +9518,7 @@ dependencies = [

 [[package]]
 name = "rustfs-notify"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -9551,7 +9552,7 @@ dependencies = [

 [[package]]
 name = "rustfs-object-capacity"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "criterion",
 "futures",
@@ -9569,7 +9570,7 @@ dependencies = [

 [[package]]
 name = "rustfs-object-data-cache"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "bytes",
 "criterion",
@@ -9585,7 +9586,7 @@ dependencies = [

 [[package]]
 name = "rustfs-obs"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "chrono",
 "crossbeam-channel",
@@ -9637,7 +9638,7 @@ dependencies = [

 [[package]]
 name = "rustfs-policy"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "base64-simd",
@@ -9666,7 +9667,7 @@ dependencies = [

 [[package]]
 name = "rustfs-protocols"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "astral-tokio-tar",
 "async-compression",
@@ -9725,7 +9726,7 @@ dependencies = [

 [[package]]
 name = "rustfs-protos"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "flatbuffers",
 "prost 0.14.4",
@@ -9747,7 +9748,7 @@ dependencies = [

 [[package]]
 name = "rustfs-replication"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "byteorder",
 "bytes",
@@ -9764,7 +9765,7 @@ dependencies = [

 [[package]]
 name = "rustfs-rio"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "arc-swap",
@@ -9802,7 +9803,7 @@ dependencies = [

 [[package]]
 name = "rustfs-rio-v2"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "aes-gcm",
 "bytes",
@@ -9824,14 +9825,14 @@ dependencies = [

 [[package]]
 name = "rustfs-s3-ops"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "rustfs-s3-types",
 ]

 [[package]]
 name = "rustfs-s3-types"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "serde",
 "serde_json",
@@ -9839,7 +9840,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3select-api"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "bytes",
@@ -9867,7 +9868,7 @@ dependencies = [

 [[package]]
 name = "rustfs-s3select-query"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-recursion",
 "async-trait",
@@ -9883,7 +9884,7 @@ dependencies = [

 [[package]]
 name = "rustfs-scanner"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "chrono",
@@ -9915,14 +9916,14 @@ dependencies = [

 [[package]]
 name = "rustfs-security-governance"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "thiserror 2.0.19",
 ]

 [[package]]
 name = "rustfs-signer"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -9938,7 +9939,7 @@ dependencies = [

 [[package]]
 name = "rustfs-storage-api"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "insta",
@@ -9952,7 +9953,7 @@ dependencies = [

 [[package]]
 name = "rustfs-targets"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "arc-swap",
 "async-nats",
@@ -10005,7 +10006,7 @@ dependencies = [

 [[package]]
 name = "rustfs-test-utils"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "rustfs-data-usage",
 "rustfs-ecstore",
@@ -10020,7 +10021,7 @@ dependencies = [

 [[package]]
 name = "rustfs-tls-runtime"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "arc-swap",
 "metrics",
@@ -10040,7 +10041,7 @@ dependencies = [

 [[package]]
 name = "rustfs-trusted-proxies"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "async-trait",
 "axum",
@@ -10076,7 +10077,7 @@ dependencies = [

 [[package]]
 name = "rustfs-utils"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "base64-simd",
 "blake2",
@@ -10097,6 +10098,7 @@ dependencies = [
 "netif",
 "proptest",
 "regex",
+ "reqwest",
 "rustix",
 "serde",
 "sha1 0.11.0",
@@ -10115,7 +10117,7 @@ dependencies = [

 [[package]]
 name = "rustfs-zip"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 dependencies = [
 "astral-tokio-tar",
 "async-compression",
@@ -69,7 +69,7 @@ edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/rustfs/rustfs"
 rust-version = "1.96.0"
-version = "1.0.0-beta.10"
+version = "1.0.0-beta.11"
 homepage = "https://rustfs.com"
 description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
 keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -86,52 +86,52 @@ redundant_clone = "warn"

 [workspace.dependencies]
 # RustFS Internal Crates
-rustfs = { path = "./rustfs", version = "1.0.0-beta.10" }
-rustfs-heal = { path = "crates/heal", version = "1.0.0-beta.10" }
-rustfs-audit = { path = "crates/audit", version = "1.0.0-beta.10" }
-rustfs-checksums = { path = "crates/checksums", version = "1.0.0-beta.10" }
-rustfs-common = { path = "crates/common", version = "1.0.0-beta.10" }
-rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-beta.10" }
-rustfs-config = { path = "./crates/config", version = "1.0.0-beta.10" }
-rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-beta.10" }
-rustfs-credentials = { path = "crates/credentials", version = "1.0.0-beta.10" }
-rustfs-crypto = { path = "crates/crypto", version = "1.0.0-beta.10" }
-rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-beta.10" }
-rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-beta.10" }
-rustfs-iam = { path = "crates/iam", version = "1.0.0-beta.10" }
-rustfs-keystone = { path = "crates/keystone", version = "1.0.0-beta.10" }
-rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-beta.10" }
-rustfs-kms = { path = "crates/kms", version = "1.0.0-beta.10" }
-rustfs-lock = { path = "crates/lock", version = "1.0.0-beta.10" }
-rustfs-madmin = { path = "crates/madmin", version = "1.0.0-beta.10" }
-rustfs-notify = { path = "crates/notify", version = "1.0.0-beta.10" }
-rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-beta.10" }
-rustfs-io-core = { path = "crates/io-core", version = "1.0.0-beta.10" }
-rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-beta.10" }
-rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.10" }
-rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-beta.10" }
-rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.10" }
-rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.10" }
-rustfs-protos = { path = "crates/protos", version = "1.0.0-beta.10" }
-rustfs-protocols = { path = "crates/protocols", version = "1.0.0-beta.10" }
-rustfs-replication = { path = "crates/replication", version = "1.0.0-beta.10" }
-rustfs-rio = { path = "crates/rio", version = "1.0.0-beta.10" }
-rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-beta.10" }
-rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-beta.10" }
-rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-beta.10" }
-rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-beta.10" }
-rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-beta.10" }
-rustfs-scanner = { path = "crates/scanner", version = "1.0.0-beta.10" }
-rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-beta.10" }
-rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-beta.10" }
-rustfs-signer = { path = "crates/signer", version = "1.0.0-beta.10" }
-rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-beta.10" }
-rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-beta.10" }
-rustfs-targets = { path = "crates/targets", version = "1.0.0-beta.10" }
-rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-beta.10" }
-rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-beta.10" }
-rustfs-utils = { path = "crates/utils", version = "1.0.0-beta.10" }
-rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.10" }
+rustfs = { path = "./rustfs", version = "1.0.0-beta.11" }
+rustfs-heal = { path = "crates/heal", version = "1.0.0-beta.11" }
+rustfs-audit = { path = "crates/audit", version = "1.0.0-beta.11" }
+rustfs-checksums = { path = "crates/checksums", version = "1.0.0-beta.11" }
+rustfs-common = { path = "crates/common", version = "1.0.0-beta.11" }
+rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-beta.11" }
+rustfs-config = { path = "./crates/config", version = "1.0.0-beta.11" }
+rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-beta.11" }
+rustfs-credentials = { path = "crates/credentials", version = "1.0.0-beta.11" }
+rustfs-crypto = { path = "crates/crypto", version = "1.0.0-beta.11" }
+rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-beta.11" }
+rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-beta.11" }
+rustfs-iam = { path = "crates/iam", version = "1.0.0-beta.11" }
+rustfs-keystone = { path = "crates/keystone", version = "1.0.0-beta.11" }
+rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-beta.11" }
+rustfs-kms = { path = "crates/kms", version = "1.0.0-beta.11" }
+rustfs-lock = { path = "crates/lock", version = "1.0.0-beta.11" }
+rustfs-madmin = { path = "crates/madmin", version = "1.0.0-beta.11" }
+rustfs-notify = { path = "crates/notify", version = "1.0.0-beta.11" }
+rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-beta.11" }
+rustfs-io-core = { path = "crates/io-core", version = "1.0.0-beta.11" }
+rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-beta.11" }
+rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.11" }
+rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-beta.11" }
+rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.11" }
+rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.11" }
+rustfs-protos = { path = "crates/protos", version = "1.0.0-beta.11" }
+rustfs-protocols = { path = "crates/protocols", version = "1.0.0-beta.11" }
+rustfs-replication = { path = "crates/replication", version = "1.0.0-beta.11" }
+rustfs-rio = { path = "crates/rio", version = "1.0.0-beta.11" }
+rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-beta.11" }
+rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-beta.11" }
+rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-beta.11" }
+rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-beta.11" }
+rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-beta.11" }
+rustfs-scanner = { path = "crates/scanner", version = "1.0.0-beta.11" }
+rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-beta.11" }
+rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-beta.11" }
+rustfs-signer = { path = "crates/signer", version = "1.0.0-beta.11" }
+rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-beta.11" }
+rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-beta.11" }
+rustfs-targets = { path = "crates/targets", version = "1.0.0-beta.11" }
+rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-beta.11" }
+rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-beta.11" }
+rustfs-utils = { path = "crates/utils", version = "1.0.0-beta.11" }
+rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.11" }

 # Async Runtime and Networking
 async-channel = "2.5.0"
@@ -116,7 +116,7 @@ chown -R 10001:10001 data logs
 docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest

 # Using specific version
-docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.10
+docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.11
 ```

 If you use [podman](https://github.com/containers/podman) instead of docker, you can install the RustFS with the below command
@@ -113,7 +113,7 @@ chown -R 10001:10001 data logs
 docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest

 # 使用指定版本运行
-docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.10
+docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.11
 ```

 如果您通过绑定挂载启用 TLS 证书目录，也请用同样方式准备该目录：
@@ -56,3 +56,33 @@ pub const DEFAULT_OBJECT_MMAP_READ_ENABLE: bool = true;
 ///
 /// Prefer [`DEFAULT_OBJECT_MMAP_READ_ENABLE`].
 pub const DEFAULT_OBJECT_ZERO_COPY_ENABLE: bool = DEFAULT_OBJECT_MMAP_READ_ENABLE;
+
+/// Environment variable capping the byte length a single mmap-copy read may
+/// materialize in memory.
+///
+/// The mmap-copy read path returns the whole requested range as one owned
+/// allocation before the first byte is served. GET/heal shard reads request
+/// the entire part span in one call, so for a large single-part object
+/// (e.g. a multi-gigabyte non-multipart upload) an uncapped mmap-copy read
+/// allocates the whole shard in memory — stalling first-byte latency past the
+/// disk-read timeout and OOM-killing memory-limited deployments
+/// (<https://github.com/rustfs/rustfs/issues/5123>). Reads longer than this
+/// cap fall back to the bounded streaming reader instead.
+///
+/// - Purpose: Bound per-shard-read memory for mmap-based reads
+/// - Acceptable values: byte count as an unsigned integer; `0` disables
+///   mmap-copy for all non-empty reads (every read streams)
+/// - Example: `export RUSTFS_OBJECT_MMAP_READ_MAX_LENGTH=8388608`
+pub const ENV_OBJECT_MMAP_READ_MAX_LENGTH: &str = "RUSTFS_OBJECT_MMAP_READ_MAX_LENGTH";
+
+/// Default mmap-copy read length cap: 32 MiB per shard read.
+///
+/// Large enough that typical multipart part shards (parts up to a few hundred
+/// megabytes across the erasure set) keep the mmap fast path, small enough
+/// that whole-part reads of huge single-part objects stream instead of
+/// materializing gigabytes per shard.
+///
+/// The cap bounds memory per shard reader, so a single part read can still
+/// materialize up to `data_shards x cap` bytes; raising the cap raises that
+/// per-request bound proportionally.
+pub const DEFAULT_OBJECT_MMAP_READ_MAX_LENGTH: usize = 32 * 1024 * 1024;
@@ -33,6 +33,7 @@ rustfs-config = { workspace = true, features = ["constants"] }
 rustfs-ecstore.workspace = true
 rustfs-data-usage.workspace = true
 rustfs-rio.workspace = true
+rustfs-utils = { workspace = true, features = ["egress"] }
 flatbuffers.workspace = true
 futures.workspace = true
 rustfs-lock.workspace = true
@@ -39,6 +39,7 @@ use local_ip_address::local_ip;
 use reqwest::StatusCode;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
+use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use s3s::Body;
 use serde_json::Value;
 use serial_test::serial;
@@ -70,6 +71,11 @@ fn target_arn(target_name: &str) -> String {
    format!("arn:rustfs:sqs:{NOTIFY_REGION}:{target_name}:webhook")
 }

+fn endpoint_origin(endpoint: &str) -> Result<String, BoxError> {
+    let parsed = reqwest::Url::parse(endpoint)?;
+    Ok(parsed.origin().ascii_serialization())
+}
+
 // ---------------------------------------------------------------------------
 // In-test HTTP event receiver
 // ---------------------------------------------------------------------------
@@ -153,7 +159,7 @@ async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver<Valu
    let endpoint_ip = local_ip()?;
    let (tx, rx) = mpsc::unbounded_channel();
    let handle = serve_event_collector(listener, tx);
-    Ok((format!("http://{endpoint_ip}.nip.io:{port}/events"), rx, handle))
+    Ok((format!("http://{}/events", std::net::SocketAddr::new(endpoint_ip, port)), rx, handle))
 }

 struct HttpsEventCollector {
@@ -209,9 +215,9 @@ fn spawn_https_event_collector(ca_path: &Path) -> Result<HttpsEventCollector, Bo
    listener.set_nonblocking(true)?;
    let addr = listener.local_addr()?;
    let endpoint_ip = local_ip()?;
-    let endpoint_host = format!("{endpoint_ip}.nip.io");
+    let endpoint_host = endpoint_ip.to_string();

-    let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![endpoint_host.clone()])?;
+    let rcgen::CertifiedKey { cert, signing_key } = rcgen::generate_simple_self_signed(vec![endpoint_host])?;
    std::fs::write(ca_path, cert.pem())?;

    let cert_chain = vec![cert.der().clone()];
@@ -246,7 +252,7 @@ fn spawn_https_event_collector(ca_path: &Path) -> Result<HttpsEventCollector, Bo
    });

    Ok(HttpsEventCollector {
-        endpoint: format!("https://{endpoint_host}:{}/events", addr.port()),
+        endpoint: format!("https://{}/events", std::net::SocketAddr::new(endpoint_ip, addr.port())),
        running,
        handle: Some(handle),
        events,
@@ -586,11 +592,17 @@ async fn test_https_webhook_target_delivers_event_with_notify_env_enabled() -> T
    init_logging();

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "true")])
-        .await?;
-
    let ca_path = Path::new(&env.temp_dir).join("https-webhook-ca.pem");
    let mut collector = spawn_https_event_collector(&ca_path)?;
+    let allowed_origin = endpoint_origin(collector.endpoint())?;
+    env.start_rustfs_server_with_env(
+        vec![],
+        &[
+            ("RUSTFS_NOTIFY_ENABLE", "true"),
+            (ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str()),
+        ],
+    )
+    .await?;
    let target = "peri1https";
    let bucket = "peri1-https-events";
    let key = "uploads/https.dat";
@@ -635,9 +647,11 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult {
    init_logging();

    let (endpoint, mut rx, handle) = spawn_event_collector().await?;
+    let allowed_origin = endpoint_origin(&endpoint)?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str())])
+        .await?;
    enable_notify_module(&env).await?;

    let bucket = "peri1-events";
@@ -789,15 +803,6 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult {
 async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
    init_logging();

-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-    enable_notify_module(&env).await?;
-
-    let bucket = "peri1-redeliver";
-    let target = "peri1redeliver";
-    let client = env.create_s3_client();
-    client.create_bucket().bucket(bucket).send().await?;
-
    // Configure the target while its endpoint is reachable so activation and
    // ARN registration complete deterministically. Registering against a dead
    // endpoint stalls behind the reachability probe's timeout and flakes the
@@ -806,10 +811,21 @@ async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
    let listener = TcpListener::bind("0.0.0.0:0").await?;
    let port = listener.local_addr()?.port();
    let endpoint_ip = local_ip()?;
-    let endpoint = format!("http://{endpoint_ip}.nip.io:{port}/events");
+    let endpoint = format!("http://{}/events", std::net::SocketAddr::new(endpoint_ip, port));
+    let allowed_origin = endpoint_origin(&endpoint)?;
    let (setup_tx, _setup_rx) = mpsc::unbounded_channel();
    let setup_handle = serve_event_collector(listener, setup_tx);

+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_OUTBOUND_ALLOW_ORIGINS, allowed_origin.as_str())])
+        .await?;
+    enable_notify_module(&env).await?;
+
+    let bucket = "peri1-redeliver";
+    let target = "peri1redeliver";
+    let client = env.create_s3_client();
+    client.create_bucket().bucket(bucket).send().await?;
+
    configure_webhook_target(&env, target, &endpoint).await?;
    wait_for_target_registered(&env, target).await?;
    put_notification_config(&client, bucket, target, "uploads/", ".dat").await?;
@@ -18,6 +18,7 @@ use http::header::{CONTENT_TYPE, HOST};
 use reqwest::StatusCode;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::{pre_sign_v4, sign_v4};
+use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use s3s::Body;
 use serial_test::serial;
 use std::collections::HashMap;
@@ -49,7 +50,7 @@ fn find_header_terminator(buf: &[u8]) -> Option<usize> {

 async fn read_http_request(
    stream: &mut tokio::net::TcpStream,
-) -> Result<(HashMap<String, String>, Vec<u8>), Box<dyn Error + Send + Sync>> {
+) -> Result<(String, HashMap<String, String>, Vec<u8>), Box<dyn Error + Send + Sync>> {
    let mut buffer = Vec::new();
    let mut chunk = [0_u8; 4096];

@@ -67,7 +68,7 @@ async fn read_http_request(
    let header_bytes = &buffer[..header_end];
    let header_text = std::str::from_utf8(header_bytes)?;
    let mut lines = header_text.split("\r\n");
-    let _request_line = lines.next().ok_or("missing request line")?;
+    let request_line = lines.next().ok_or("missing request line")?.to_string();
    let mut headers = HashMap::new();
    for line in lines {
        if line.is_empty() {
@@ -79,8 +80,9 @@ async fn read_http_request(

    let content_length = headers
        .get("content-length")
-        .ok_or("missing content-length header")?
-        .parse::<usize>()?;
+        .map(|value| value.parse::<usize>())
+        .transpose()?
+        .unwrap_or_default();
    let body_offset = header_end + 4;
    while buffer.len().saturating_sub(body_offset) < content_length {
        let read = stream.read(&mut chunk).await?;
@@ -90,7 +92,7 @@ async fn read_http_request(
        buffer.extend_from_slice(&chunk[..read]);
    }

-    Ok((headers, buffer[body_offset..body_offset + content_length].to_vec()))
+    Ok((request_line, headers, buffer[body_offset..body_offset + content_length].to_vec()))
 }

 async fn spawn_object_lambda_webhook_server() -> Result<
@@ -130,9 +132,17 @@ async fn spawn_object_lambda_webhook_server_with_response(
    let handle = tokio::spawn(async move {
        loop {
            let (mut stream, _) = listener.accept().await?;
-            let Ok(Ok((headers, body))) = timeout(Duration::from_secs(2), read_http_request(&mut stream)).await else {
+            let Ok(Ok((request_line, headers, body))) = timeout(Duration::from_secs(2), read_http_request(&mut stream)).await
+            else {
                continue;
            };
+            if request_line == "HEAD / HTTP/1.1" {
+                stream
+                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
+                    .await?;
+                stream.shutdown().await?;
+                continue;
+            }
            let payload: serde_json::Value = serde_json::from_slice(&body)?;

            let output_route = payload["getObjectContext"]["outputRoute"]
@@ -412,9 +422,33 @@ async fn wait_for_target_absence(
    Err(format!("target {target_name} remained visible in admin APIs; targets={last_targets}, arns={last_arns:?}").into())
 }

-async fn restart_rustfs_server(env: &mut RustFSTestEnvironment) -> Result<(), Box<dyn Error + Send + Sync>> {
+fn endpoint_origin(endpoint: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
+    Ok(reqwest::Url::parse(endpoint)?.origin().ascii_serialization())
+}
+
+async fn start_rustfs_server_for_endpoint(
+    env: &mut RustFSTestEnvironment,
+    endpoint: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let origin = endpoint_origin(endpoint)?;
+    env.start_rustfs_server_with_env(
+        vec![],
+        &[
+            (ENV_OUTBOUND_ALLOW_ORIGINS, origin.as_str()),
+            ("RUSTFS_NOTIFY_ENABLE", "true"),
+        ],
+    )
+    .await
+}
+
+async fn restart_rustfs_server(env: &mut RustFSTestEnvironment, endpoint: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let origin = endpoint_origin(endpoint)?;
    env.stop_server();
-    env.start_rustfs_server_without_cleanup(vec![]).await
+    env.start_rustfs_server_without_cleanup_with_env(&[
+        (ENV_OUTBOUND_ALLOW_ORIGINS, origin.as_str()),
+        ("RUSTFS_NOTIFY_ENABLE", "true"),
+    ])
+    .await
 }

 async fn spawn_http_origin_probe_server() -> Result<
@@ -521,7 +555,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result
    let (webhook_url, _request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let target_name = "restart-target";
    configure_webhook_target(&env, target_name, &webhook_url, "secret-token").await?;
@@ -535,7 +569,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result
        "target ARN missing after initial configure: {visible_arns:?}"
    );

-    restart_rustfs_server(&mut env).await?;
+    restart_rustfs_server(&mut env, &webhook_url).await?;

    let (targets_after_restart, arns_after_restart) = wait_for_target_visibility(&env, target_name).await?;
    assert!(notification_target_is_listed(&targets_after_restart, target_name));
@@ -556,7 +590,7 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result
        "target ARN still visible after delete: {arns_after_delete:?}"
    );

-    restart_rustfs_server(&mut env).await?;
+    restart_rustfs_server(&mut env, &webhook_url).await?;

    let (targets_after_delete_restart, arns_after_delete_restart) = wait_for_target_absence(&env, target_name).await?;
    assert!(!notification_target_is_listed(&targets_after_delete_restart, target_name));
@@ -581,8 +615,7 @@ async fn test_notification_target_with_path_is_online_via_transport_probe() -> R
    let (webhook_url, mut probe_rx, probe_handle) = spawn_http_origin_probe_server().await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "true")])
-        .await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let target_name = "path-probe";
    configure_webhook_target(&env, target_name, &webhook_url, "secret-token").await?;
@@ -615,7 +648,7 @@ async fn test_get_object_lambda_accepts_presigned_requests() -> Result<(), Box<d
    let (webhook_url, request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e-presigned";
    let key = "input.txt";
@@ -656,7 +689,7 @@ async fn test_get_object_lambda_accepts_named_webhook_target_arn() -> Result<(),
    let (webhook_url, request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e-named-target";
    let key = "input.txt";
@@ -696,7 +729,7 @@ async fn test_get_object_lambda_invokes_runtime_webhook_target() -> Result<(), B
    let (webhook_url, request_rx, webhook_handle) = spawn_object_lambda_webhook_server().await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e";
    let key = "input.txt";
@@ -777,7 +810,7 @@ async fn test_get_object_lambda_passthroughs_non_success_webhook_response() -> R
    .await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e-failure";
    let key = "input.txt";
@@ -832,7 +865,7 @@ async fn test_get_object_lambda_rejects_success_response_without_auth_headers()
    .await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e-missing-auth";
    let key = "input.txt";
@@ -879,7 +912,7 @@ async fn test_get_object_lambda_rejects_success_response_with_mismatched_auth_he
    .await?;

    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    start_rustfs_server_for_endpoint(&mut env, &webhook_url).await?;

    let bucket = "object-lambda-e2e-mismatched-auth";
    let key = "input.txt";
@@ -4048,17 +4048,23 @@ async fn test_site_replication_allows_private_ca_https_with_ca_cert_pem_real_dua

 #[tokio::test]
 #[serial]
-async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> Result<(), Box<dyn Error + Send + Sync>> {
+async fn test_site_replication_resync_lifecycle_survives_real_server_restart() -> Result<(), Box<dyn Error + Send + Sync>> {
    init_logging();
+    let resync_process_env = [
+        ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true"),
+        // Verbose server logging can block startup when this focused test is run
+        // through a captured test process rather than nextest.
+        ("RUST_LOG", "error"),
+    ];

    let mut source_env = RustFSTestEnvironment::new().await?;
-    source_env
-        .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
-        .await?;
+    source_env.capture_log_path = Some(format!("{}/server.log", source_env.temp_dir));
+    source_env.start_rustfs_server_with_env(vec![], &resync_process_env).await?;

    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.capture_log_path = Some(format!("{}/server.log", target_env.temp_dir));
    target_env
-        .start_rustfs_server_without_cleanup_with_env(LOOPBACK_REPLICATION_TARGET_ENV)
+        .start_rustfs_server_without_cleanup_with_env(&resync_process_env)
        .await?;

    let source_bucket = "site-repl-resync-src";
@@ -4106,12 +4112,12 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
    wait_for_bucket_on_target(&source_client, source_bucket).await?;
    let target_arn = wait_for_remote_target_arn(&source_env, source_bucket).await?;

-    for idx in 0..32 {
+    for idx in 0..96 {
        source_client
            .put_object()
            .bucket(source_bucket)
            .key(format!("resync-object-{idx:02}"))
-            .body(ByteStream::from(vec![b'x'; 256 * 1024]))
+            .body(ByteStream::from(vec![b'x'; 512 * 1024]))
            .send()
            .await?;
    }
@@ -4119,18 +4125,31 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
    let started = site_replication_resync_op(&source_env, "start", &remote_peer).await?;
    assert_eq!(started.status, "success", "unexpected start result: {:?}", started);
    assert!(
-        started
-            .buckets
-            .iter()
-            .any(|bucket| bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "success")),
+        started.buckets.iter().any(|bucket| {
+            bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "running" | "completed" | "success")
+        }),
        "source bucket start status missing: {:?}",
        started
    );
    assert!(!started.resync_id.is_empty(), "start response omitted the resync id: {:?}", started);
    let started_reset_id = started.resync_id.clone();

+    assert!(
+        matches!(started.state.as_str(), "pending" | "running"),
+        "the fixture must keep the first generation active long enough to test duplicate start: {:?}",
+        started
+    );
+    let duplicate_err = site_replication_resync_op(&source_env, "start", &remote_peer)
+        .await
+        .expect_err("duplicate start must be rejected while a generation is active");
+    assert!(
+        duplicate_err.to_string().contains("already active"),
+        "unexpected duplicate start error: {duplicate_err}"
+    );
+
    let canceled = site_replication_resync_op(&source_env, "cancel", &remote_peer).await?;
    assert_eq!(canceled.status, "success", "unexpected cancel result: {:?}", canceled);
+    assert_eq!(canceled.state, "canceled");
    assert!(
        canceled
            .buckets
@@ -4139,34 +4158,56 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
        "source bucket cancel status missing: {:?}",
        canceled
    );
+    let canceled_again = site_replication_resync_op(&source_env, "cancel", &remote_peer).await?;
+    assert_eq!(canceled_again.resync_id, canceled.resync_id, "repeated cancel must be idempotent");
+    assert_eq!(canceled_again.state, "canceled");

    let canceled_target =
        wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |target| target.status == "Canceled").await?;
-    assert_eq!(canceled_target.status, "Canceled");
    assert_eq!(canceled_target.reset_id, started_reset_id);

    let restarted = site_replication_resync_op(&source_env, "start", &remote_peer).await?;
    assert_eq!(restarted.status, "success", "unexpected restart result: {:?}", restarted);
+    assert_ne!(restarted.resync_id, started_reset_id);
    assert!(
+        matches!(restarted.state.as_str(), "pending" | "running"),
+        "the second generation must be active before the process restart: {:?}",
        restarted
-            .buckets
-            .iter()
-            .any(|bucket| bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "success")),
-        "source bucket restart status missing: {:?}",
-        restarted
+    );
+    let restarted_reset_id = restarted.resync_id.clone();
+
+    source_env.restart_server_preserving_data(vec![], &resync_process_env).await?;
+    wait_for_site_replication_enabled(&source_env, 2).await?;
+
+    let after_restart = site_replication_resync_op(&source_env, "status", &remote_peer).await?;
+    assert_eq!(
+        after_restart.resync_id, restarted_reset_id,
+        "server restart changed the durable resync id"
+    );
+    assert_eq!(after_restart.generation, restarted.generation);
+    assert_eq!(after_restart.created_at, restarted.created_at);
+    assert!(
+        matches!(after_restart.state.as_str(), "pending" | "running" | "completed" | "failed"),
+        "unexpected recovered lifecycle state: {:?}",
+        after_restart
+    );
+    assert!(
+        after_restart.buckets.iter().any(|bucket| bucket.bucket == source_bucket),
+        "durable status lost the source bucket after restart: {:?}",
+        after_restart
    );
    let restart_snapshot = get_replication_reset_status(&source_env, source_bucket, &target_arn).await?;
    let restarted_target = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |target| {
-        !target.reset_id.is_empty() && target.reset_id != started_reset_id
+        target.reset_id == restarted_reset_id
    })
    .await
    .map_err(|err| {
        format!(
            "restart ids: start={} restart={} snapshot={:?}; {err}",
-            started_reset_id, restarted.resync_id, restart_snapshot.targets
+            started_reset_id, restarted_reset_id, restart_snapshot.targets
        )
    })?;
-    assert_ne!(restarted_target.reset_id, started_reset_id);
+    assert_eq!(restarted_target.reset_id, restarted_reset_id);

    Ok(())
 }
@@ -219,31 +219,62 @@ impl PeerRestClient {
            recovery_running: Arc::new(AtomicBool::new(false)),
        }
    }
-    pub async fn new_clients(eps: EndpointServerPools) -> (Vec<Option<Self>>, Vec<Option<Self>>) {
-        if !runtime_sources::setup_is_dist_erasure().await {
-            return (Vec::new(), Vec::new());
-        }

-        let eps = eps.clone();
-        let hosts = eps.hosts_sorted();
-        let mut remote = Vec::with_capacity(hosts.len());
-        let mut all = vec![None; hosts.len()];
-        for (i, hs_host) in hosts.iter().enumerate() {
-            if let Some(host) = hs_host
-                && let Some(grid_host) = eps.find_grid_hosts_from_peer(host)
-            {
-                let client = PeerRestClient::new(host.clone(), grid_host);
+    fn build_clients_from_slots(
+        slots: Vec<(String, Option<String>, bool)>,
+    ) -> (Vec<Option<Self>>, Vec<Option<Self>>, Vec<String>) {
+        let mut remote = Vec::with_capacity(slots.len().saturating_sub(1));
+        let mut all = vec![None; slots.len()];
+        let mut remote_topology_hosts = Vec::with_capacity(slots.len().saturating_sub(1));

-                all[i] = Some(client.clone());
-                remote.push(Some(client));
+        for (idx, (peer_host_port, grid_host, is_local)) in slots.into_iter().enumerate() {
+            if is_local {
+                continue;
            }
+
+            let client = match grid_host {
+                Some(grid_host) => match XHost::try_from(peer_host_port.clone()) {
+                    Ok(host) => Some(PeerRestClient::new(host, grid_host)),
+                    Err(err) => {
+                        warn!(peer = %peer_host_port, "Xhost parse failed while constructing peer client: {err:?}");
+                        None
+                    }
+                },
+                None => {
+                    warn!(peer = %peer_host_port, "grid host is missing while constructing peer client");
+                    None
+                }
+            };
+
+            all[idx] = client.clone();
+            remote.push(client);
+            remote_topology_hosts.push(peer_host_port);
        }

+        (remote, all, remote_topology_hosts)
+    }
+
+    pub async fn new_clients(eps: EndpointServerPools) -> (Vec<Option<Self>>, Vec<Option<Self>>) {
+        let (remote, all, _) = Self::new_clients_with_topology(eps).await;
+        (remote, all)
+    }
+
+    pub async fn new_clients_with_topology(eps: EndpointServerPools) -> (Vec<Option<Self>>, Vec<Option<Self>>, Vec<String>) {
+        if !runtime_sources::setup_is_dist_erasure().await {
+            return (Vec::new(), Vec::new(), Vec::new());
+        }
+
+        let (remote, all, remote_topology_hosts) = Self::build_clients_from_slots(eps.peer_grid_host_slots_sorted());
+
        if all.len() != remote.len() + 1 {
-            warn!("Expected number of all hosts ({}) to be remote +1 ({})", all.len(), remote.len());
+            warn!(
+                all_hosts = all.len(),
+                remote_slots = remote.len(),
+                "Expected number of all hosts to be remote slots + local node"
+            );
        }

-        (remote, all)
+        (remote, all, remote_topology_hosts)
    }

    pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
@@ -1536,6 +1567,36 @@ mod tests {
        )
    }

+    #[test]
+    fn build_clients_from_slots_preserves_missing_remote_topology_slots() {
+        let slots = vec![
+            ("127.0.0.1:9000".to_string(), None, true),
+            ("127.0.0.1:9001".to_string(), Some("http://127.0.0.1:9001".to_string()), false),
+            ("127.0.0.1:notaport".to_string(), Some("http://127.0.0.1:notaport".to_string()), false),
+            ("127.0.0.1:9003".to_string(), None, false),
+        ];
+
+        let (remote, all, remote_topology_hosts) = PeerRestClient::build_clients_from_slots(slots);
+
+        assert_eq!(remote.len(), 3, "local node is excluded but remote slots are not compacted away");
+        assert_eq!(all.len(), 4, "all slots preserve the sorted cluster topology shape");
+        assert_eq!(
+            remote_topology_hosts,
+            vec![
+                "127.0.0.1:9001".to_string(),
+                "127.0.0.1:notaport".to_string(),
+                "127.0.0.1:9003".to_string()
+            ]
+        );
+        assert!(remote[0].is_some(), "valid remote peer should get a client");
+        assert!(remote[1].is_none(), "unparseable remote peer should remain observable as a missing slot");
+        assert!(remote[2].is_none(), "missing grid host should remain observable as a missing slot");
+        assert!(all[0].is_none(), "local node is represented by the local server_info row");
+        assert!(all[1].is_some());
+        assert!(all[2].is_none());
+        assert!(all[3].is_none());
+    }
+
    #[test]
    fn scanner_activity_requires_restart_safe_peer_identity() {
        let missing_instance = ScannerActivityResponse {
@@ -15,7 +15,11 @@
 use crate::cluster::rpc::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
 use crate::data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, load_data_usage_from_backend_cached};
 use crate::error::{Error, Result};
-use crate::{disk::endpoint::Endpoint, runtime::sources as runtime_sources};
+use crate::{
+    disk::endpoint::{Endpoint, EndpointType},
+    layout::endpoints::EndpointServerPools,
+    runtime::sources as runtime_sources,
+};

 use crate::data_usage::load_data_usage_cache;
 use crate::storage_api_contracts::admin::StorageAdminApi;
@@ -291,6 +295,20 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage {
        let after4 = OffsetDateTime::now_utc();

        warn!("backend_info end {:?}", after4 - after3);
+        if let Some(endpoints) = runtime_sources::endpoint_pools() {
+            let (added, report) = reconcile_servers_with_endpoint_topology(&mut servers, &endpoints);
+            if added > 0 || !report.is_complete() {
+                warn!(
+                    event = "admin_v3_info_topology_incomplete",
+                    synthesized_servers = added,
+                    expected_drives = report.expected_drives,
+                    observed_drives = report.observed_drives,
+                    missing_drives = report.missing_drive_ids.len(),
+                    duplicate_drives = report.duplicate_drive_ids.len(),
+                    "admin v3 server_info reconciled endpoint topology before computing backend counters"
+                );
+            }
+        }

        let mut all_disks: Vec<Disk> = Vec::new();
        for server in servers.iter() {
@@ -397,6 +415,228 @@ fn get_online_offline_disks_stats(disks_info: &[Disk]) -> (BackendDisks, Backend
    (BackendDisks(online_disks), BackendDisks(offline_disks), BackendDisks(unknown_disks))
 }

+#[derive(Debug, Default, PartialEq, Eq)]
+struct TopologyCompletenessReport {
+    expected_drives: usize,
+    observed_drives: usize,
+    missing_drive_ids: Vec<TopologyDriveKey>,
+    duplicate_drive_ids: Vec<TopologyDriveKey>,
+}
+
+impl TopologyCompletenessReport {
+    fn is_complete(&self) -> bool {
+        self.expected_drives == self.observed_drives && self.missing_drive_ids.is_empty() && self.duplicate_drive_ids.is_empty()
+    }
+}
+
+#[derive(Debug)]
+struct TopologyMember {
+    display_endpoint: String,
+    disks: Vec<Disk>,
+    drive_keys: Vec<TopologyDriveKey>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+struct TopologyDriveKey {
+    pool_index: i32,
+    set_index: i32,
+    disk_index: i32,
+    member_index: usize,
+}
+
+#[derive(Debug, Default)]
+struct EndpointTopology {
+    members: Vec<TopologyMember>,
+    aliases: HashMap<String, usize>,
+    expected_drive_ids: HashSet<TopologyDriveKey>,
+}
+
+impl EndpointTopology {
+    fn from_endpoint_pools(endpoints: &EndpointServerPools) -> Self {
+        let mut topology = Self::default();
+        let mut by_host_port = HashMap::new();
+        let mut display_aliases: HashMap<String, Option<usize>> = HashMap::new();
+
+        for pool in endpoints.as_ref() {
+            for ep in pool.endpoints.as_ref() {
+                if ep.get_type() != EndpointType::Url {
+                    continue;
+                }
+                let host_port = ep.host_port();
+                if host_port.is_empty() {
+                    continue;
+                }
+
+                let member_index = match by_host_port.get(&host_port).copied() {
+                    Some(index) => index,
+                    None => {
+                        let display_endpoint = ep.url.host_str().map(str::to_owned).unwrap_or_else(|| host_port.clone());
+                        let index = topology.members.len();
+                        topology.members.push(TopologyMember {
+                            display_endpoint,
+                            disks: Vec::new(),
+                            drive_keys: Vec::new(),
+                        });
+                        by_host_port.insert(host_port.clone(), index);
+                        index
+                    }
+                };
+
+                topology.aliases.entry(host_port.clone()).or_insert(member_index);
+                topology.aliases.entry(ep.to_string()).or_insert(member_index);
+                if let Some(display_endpoint) = ep.url.host_str() {
+                    match display_aliases.entry(display_endpoint.to_owned()) {
+                        std::collections::hash_map::Entry::Vacant(entry) => {
+                            entry.insert(Some(member_index));
+                        }
+                        std::collections::hash_map::Entry::Occupied(mut entry) => {
+                            if entry.get().is_some_and(|index| index != member_index) {
+                                entry.insert(None);
+                            }
+                        }
+                    }
+                }
+
+                let drive_key = TopologyDriveKey {
+                    pool_index: ep.pool_idx,
+                    set_index: ep.set_idx,
+                    disk_index: ep.disk_idx,
+                    member_index,
+                };
+                topology.expected_drive_ids.insert(drive_key.clone());
+                topology.members[member_index].drive_keys.push(drive_key);
+                topology.members[member_index].disks.push(Disk {
+                    endpoint: ep.to_string(),
+                    state: ITEM_UNKNOWN.to_string(),
+                    pool_index: ep.pool_idx,
+                    set_index: ep.set_idx,
+                    disk_index: ep.disk_idx,
+                    ..Default::default()
+                });
+            }
+        }
+
+        for (display_endpoint, member_index) in display_aliases {
+            if let Some(member_index) = member_index {
+                topology.aliases.entry(display_endpoint).or_insert(member_index);
+            }
+        }
+
+        topology
+    }
+
+    fn is_empty(&self) -> bool {
+        self.members.is_empty()
+    }
+
+    fn observe_servers(&self, servers: &[ServerProperties]) -> (Vec<bool>, HashMap<TopologyDriveKey, usize>) {
+        let mut observed_members = vec![false; self.members.len()];
+        let mut observed_drive_counts = HashMap::with_capacity(self.expected_drive_ids.len());
+
+        for server in servers {
+            if let Some(member_index) = self.member_index_from_endpoint(&server.endpoint) {
+                observed_members[member_index] = true;
+            }
+
+            for disk in &server.disks {
+                if let Some(member_index) = self.member_index_from_endpoint(&disk.endpoint) {
+                    observed_members[member_index] = true;
+                    let drive_key = TopologyDriveKey {
+                        pool_index: disk.pool_index,
+                        set_index: disk.set_index,
+                        disk_index: disk.disk_index,
+                        member_index,
+                    };
+                    if self.expected_drive_ids.contains(&drive_key) {
+                        *observed_drive_counts.entry(drive_key).or_insert(0) += 1;
+                    }
+                }
+            }
+        }
+
+        (observed_members, observed_drive_counts)
+    }
+
+    fn member_index_from_endpoint(&self, endpoint: &str) -> Option<usize> {
+        if let Some(index) = self.aliases.get(endpoint) {
+            return Some(*index);
+        }
+
+        Endpoint::try_from(endpoint)
+            .ok()
+            .and_then(|ep| self.aliases.get(&ep.host_port()).copied())
+    }
+
+    fn report_from_counts(&self, observed_drive_counts: HashMap<TopologyDriveKey, usize>) -> TopologyCompletenessReport {
+        let mut missing_drive_ids = Vec::new();
+        let mut duplicate_drive_ids = Vec::new();
+        for id in &self.expected_drive_ids {
+            match observed_drive_counts.get(id).copied().unwrap_or(0) {
+                0 => missing_drive_ids.push(id.clone()),
+                1 => {}
+                _ => duplicate_drive_ids.push(id.clone()),
+            }
+        }
+        missing_drive_ids.sort();
+        duplicate_drive_ids.sort();
+
+        TopologyCompletenessReport {
+            expected_drives: self.expected_drive_ids.len(),
+            observed_drives: observed_drive_counts.values().sum(),
+            missing_drive_ids,
+            duplicate_drive_ids,
+        }
+    }
+}
+
+fn reconcile_servers_with_endpoint_topology(
+    servers: &mut Vec<ServerProperties>,
+    endpoints: &EndpointServerPools,
+) -> (usize, TopologyCompletenessReport) {
+    let topology = EndpointTopology::from_endpoint_pools(endpoints);
+    if topology.is_empty() {
+        return (0, TopologyCompletenessReport::default());
+    }
+
+    let (observed_members, mut observed_drive_counts) = topology.observe_servers(servers);
+    let mut missing: Vec<_> = topology
+        .members
+        .iter()
+        .enumerate()
+        .filter(|(index, _)| !observed_members[*index])
+        .map(|(_, member)| {
+            for drive_key in &member.drive_keys {
+                *observed_drive_counts.entry(drive_key.clone()).or_insert(0) += 1;
+            }
+            ServerProperties {
+                endpoint: member.display_endpoint.clone(),
+                state: ITEM_UNKNOWN.to_string(),
+                disks: member.disks.clone(),
+                ..Default::default()
+            }
+        })
+        .collect();
+    missing.sort_by(|a, b| a.endpoint.cmp(&b.endpoint));
+
+    let added = missing.len();
+    servers.extend(missing);
+    let report = topology.report_from_counts(observed_drive_counts);
+
+    (added, report)
+}
+
+fn server_topology_completeness_report(
+    servers: &[ServerProperties],
+    endpoints: &EndpointServerPools,
+) -> TopologyCompletenessReport {
+    let topology = EndpointTopology::from_endpoint_pools(endpoints);
+    if topology.is_empty() {
+        return TopologyCompletenessReport::default();
+    }
+    let (_, observed_drive_counts) = topology.observe_servers(servers);
+    topology.report_from_counts(observed_drive_counts)
+}
+
 async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>> {
    let Some(store) = runtime_sources::object_store_handle() else {
        return Err(Error::other("ServerNotInitialized"));
@@ -449,12 +689,16 @@ pub fn get_commit_id() -> String {
 mod tests {
    use serial_test::serial;

+    use crate::layout::{
+        endpoint::Endpoint,
+        endpoints::{EndpointServerPools, Endpoints, PoolEndpoints},
+    };
    use crate::runtime::sources as runtime_sources;
-    use rustfs_madmin::{Disk, ITEM_OFFLINE, ITEM_UNKNOWN};
+    use rustfs_madmin::{Disk, ITEM_OFFLINE, ITEM_ONLINE, ITEM_UNKNOWN, ServerProperties};

    use super::{
        DATA_USAGE_UNAVAILABLE_ERROR, apply_data_usage_result, get_local_server_property, get_online_offline_disks_stats,
-        get_server_info,
+        get_server_info, reconcile_servers_with_endpoint_topology, server_topology_completeness_report,
    };

    fn disk_with_state(endpoint: &str, state: &str) -> Disk {
@@ -465,6 +709,50 @@ mod tests {
        }
    }

+    fn topology_endpoint(host: &str, pool_index: usize, set_index: usize, disk_index: usize) -> Endpoint {
+        topology_endpoint_url(format!("http://{host}:9000/data{disk_index}").as_str(), pool_index, set_index, disk_index)
+    }
+
+    fn topology_endpoint_url(url: &str, pool_index: usize, set_index: usize, disk_index: usize) -> Endpoint {
+        let mut endpoint = Endpoint::try_from(url).expect("URL endpoint should parse");
+        endpoint.set_pool_index(pool_index);
+        endpoint.set_set_index(set_index);
+        endpoint.set_disk_index(disk_index);
+        endpoint
+    }
+
+    fn topology_with_hosts(hosts: &[&str]) -> EndpointServerPools {
+        let endpoints: Vec<Endpoint> = hosts
+            .iter()
+            .enumerate()
+            .map(|(disk_index, host)| topology_endpoint(host, 0, 0, disk_index))
+            .collect();
+        EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: hosts.len(),
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: String::new(),
+            platform: String::new(),
+        }])
+    }
+
+    fn server_with_disk(host: &str, disk_index: i32, state: &str) -> ServerProperties {
+        ServerProperties {
+            endpoint: host.to_string(),
+            state: ITEM_ONLINE.to_string(),
+            disks: vec![Disk {
+                endpoint: format!("http://{host}:9000/data{disk_index}"),
+                state: state.to_string(),
+                pool_index: 0,
+                set_index: 0,
+                disk_index,
+                ..Default::default()
+            }],
+            ..Default::default()
+        }
+    }
+
    #[test]
    fn disk_stats_split_unknown_into_its_own_bucket() {
        // A member whose properties RPC could not be answered contributes
@@ -493,6 +781,93 @@ mod tests {
        );
    }

+    #[test]
+    fn topology_reconcile_synthesizes_missing_member_as_unknown() {
+        let endpoints = topology_with_hosts(&["rustfs-1", "rustfs-2", "rustfs-3", "rustfs-4"]);
+        let mut servers = vec![
+            server_with_disk("rustfs-1", 0, "ok"),
+            server_with_disk("rustfs-2", 1, "ok"),
+            server_with_disk("rustfs-3", 2, "ok"),
+        ];
+
+        let (added, report) = reconcile_servers_with_endpoint_topology(&mut servers, &endpoints);
+        let (_, _, unknown) =
+            get_online_offline_disks_stats(&servers.iter().flat_map(|server| server.disks.clone()).collect::<Vec<_>>());
+
+        assert_eq!(added, 1, "the missing fourth topology member must be synthesized");
+        assert!(report.is_complete(), "synthesized drives should complete the topology report");
+        assert_eq!(servers.len(), 4, "v3 server list must preserve topology membership length");
+        let synthesized = servers
+            .iter()
+            .find(|server| server.endpoint == "rustfs-4")
+            .expect("missing member should be present");
+        assert_eq!(synthesized.state, ITEM_UNKNOWN);
+        assert_eq!(synthesized.disks.len(), 1);
+        assert_eq!(synthesized.disks[0].endpoint, "http://rustfs-4:9000/data3");
+        assert_eq!(synthesized.disks[0].disk_index, 3);
+        assert_eq!(unknown.sum(), 1, "the synthesized drive must land in unknownDisks");
+    }
+
+    #[test]
+    fn topology_reconcile_does_not_duplicate_existing_synthesized_rows() {
+        let endpoints = topology_with_hosts(&["rustfs-1", "rustfs-2"]);
+        let mut servers = vec![
+            server_with_disk("rustfs-1", 0, "ok"),
+            server_with_disk("rustfs-2", 1, ITEM_UNKNOWN),
+        ];
+        servers[1].state = ITEM_UNKNOWN.to_string();
+
+        let (added, report) = reconcile_servers_with_endpoint_topology(&mut servers, &endpoints);
+
+        assert_eq!(
+            added, 0,
+            "an existing unknown/degraded/offline row with topology drives already represents the member"
+        );
+        assert!(report.is_complete(), "existing synthesized rows should already complete the topology");
+        assert_eq!(servers.len(), 2);
+    }
+
+    #[test]
+    fn topology_reconcile_does_not_over_match_ambiguous_host_only_aliases() {
+        let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(vec![
+                topology_endpoint_url("http://rustfs-1:9000/data0", 0, 0, 0),
+                topology_endpoint_url("http://rustfs-1:9001/data1", 0, 0, 1),
+            ]),
+            cmd_line: String::new(),
+            platform: String::new(),
+        }]);
+        let mut servers = vec![server_with_disk("rustfs-1", 0, "ok")];
+
+        let (added, report) = reconcile_servers_with_endpoint_topology(&mut servers, &endpoints);
+
+        assert_eq!(added, 1, "host-only aliases are ambiguous when one host has multiple topology ports");
+        assert!(report.is_complete());
+        assert!(
+            servers
+                .iter()
+                .any(|server| server.disks.iter().any(|disk| disk.endpoint == "http://rustfs-1:9001/data1")),
+            "the second host:port member should be synthesized from exact topology"
+        );
+    }
+
+    #[test]
+    fn topology_report_detects_duplicate_drive_identity_with_balanced_total() {
+        let endpoints = topology_with_hosts(&["rustfs-1", "rustfs-2"]);
+        let servers = vec![server_with_disk("rustfs-1", 0, "ok"), server_with_disk("rustfs-1", 0, "ok")];
+
+        let report = server_topology_completeness_report(&servers, &endpoints);
+
+        assert_eq!(report.expected_drives, 2);
+        assert_eq!(report.observed_drives, 2, "a plain total-count check would look balanced");
+        assert_eq!(report.missing_drive_ids.len(), 1, "rustfs-2's drive identity is absent");
+        assert_eq!(report.duplicate_drive_ids.len(), 1, "rustfs-1's drive identity is duplicated");
+        assert!(!report.is_complete());
+    }
+
    #[test]
    fn data_usage_errors_are_sanitized_in_server_info() {
        let mut buckets = rustfs_madmin::Buckets::default();
@@ -22,7 +22,10 @@ use crate::diagnostics::get::{
 use crate::disk::{self, DiskAPI as _, DiskStore, FileReader, MmapCopyStageMetrics, error::DiskError};
 use crate::erasure::coding::{BitrotReader, BitrotWriterWrapper, CustomWriter};
 use bytes::Bytes;
-use rustfs_config::{DEFAULT_OBJECT_MMAP_READ_ENABLE, ENV_OBJECT_MMAP_READ_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
+use rustfs_config::{
+    DEFAULT_OBJECT_MMAP_READ_ENABLE, DEFAULT_OBJECT_MMAP_READ_MAX_LENGTH, ENV_OBJECT_MMAP_READ_ENABLE,
+    ENV_OBJECT_MMAP_READ_MAX_LENGTH, ENV_OBJECT_ZERO_COPY_ENABLE,
+};
 use rustfs_utils::HashAlgorithm;
 use std::future::Future;
 use std::io::{self, Cursor};
@@ -81,6 +84,21 @@ pub(crate) fn object_mmap_read_enabled() -> bool {
    )
 }

+/// Mmap-copy read length cap. Cached: this is consulted on every shard open
+/// and `std::env::var` takes a process-global lock. In test builds the env
+/// var is read directly so `temp_env` overrides take effect.
+pub(crate) fn object_mmap_read_max_length() -> usize {
+    #[cfg(test)]
+    {
+        rustfs_utils::get_env_usize(ENV_OBJECT_MMAP_READ_MAX_LENGTH, DEFAULT_OBJECT_MMAP_READ_MAX_LENGTH)
+    }
+    #[cfg(not(test))]
+    {
+        static CACHED: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+        *CACHED.get_or_init(|| rustfs_utils::get_env_usize(ENV_OBJECT_MMAP_READ_MAX_LENGTH, DEFAULT_OBJECT_MMAP_READ_MAX_LENGTH))
+    }
+}
+
 #[derive(Clone)]
 struct BitrotReaderSource {
    inline_data: Option<Bytes>,
@@ -305,7 +323,28 @@ async fn open_disk_reader(
    let metrics_path = metrics_path.filter(|_| rustfs_io_metrics::get_stage_metrics_enabled());
    let stage_metrics_enabled = metrics_path.is_some();

-    if use_mmap_read && disk.is_local() {
+    // Mmap-copy materializes the whole `offset..offset+length` range as one
+    // owned allocation before any byte is served, and GET/heal shard reads
+    // request the entire part span in one call. Over-cap reads (e.g. a huge
+    // single-part object, rustfs#5123) take the bounded streaming path below.
+    let use_mmap_copy = if use_mmap_read && disk.is_local() {
+        let mmap_read_cap = object_mmap_read_max_length();
+        let within_cap = length <= mmap_read_cap;
+        if !within_cap {
+            rustfs_io_metrics::record_zero_copy_fallback("read_length_exceeds_cap");
+            debug!(
+                length,
+                mmap_read_cap,
+                path = %path,
+                "zero_copy_read_skipped_over_cap"
+            );
+        }
+        within_cap
+    } else {
+        false
+    };
+
+    if use_mmap_copy {
        let start = stage_metrics_enabled.then(Instant::now);
        let zero_copy_start = Instant::now();
        let mmap_metrics = metrics_path.map(|metrics_path| MmapCopyStageMetrics {
@@ -685,6 +724,96 @@ mod tests {
        );
    }

+    #[test]
+    fn object_mmap_read_max_length_defaults_and_env_override() {
+        temp_env::with_var(ENV_OBJECT_MMAP_READ_MAX_LENGTH, None::<&str>, || {
+            assert_eq!(object_mmap_read_max_length(), DEFAULT_OBJECT_MMAP_READ_MAX_LENGTH);
+        });
+        temp_env::with_var(ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("1024"), || {
+            assert_eq!(object_mmap_read_max_length(), 1024);
+        });
+        temp_env::with_var(ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("0"), || {
+            assert_eq!(object_mmap_read_max_length(), 0);
+        });
+    }
+
+    // rustfs#5123: whole-part shard reads of large single-part objects must not
+    // be materialized in memory by the mmap-copy path; over-cap reads stream.
+    #[tokio::test]
+    async fn open_disk_reader_streams_when_length_exceeds_mmap_cap() {
+        use crate::disk::endpoint::Endpoint;
+        use crate::disk::{DiskOption, new_disk};
+        use tokio::io::AsyncReadExt;
+
+        let dir = tempfile::tempdir().expect("tempdir should be created");
+        let mut endpoint =
+            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("local disk should be created");
+
+        let payload = vec![7u8; 4096];
+        disk.make_volume("test-bucket").await.expect("volume should be created");
+        disk.write_all("test-bucket", "obj/part.1", Bytes::from(payload.clone()))
+            .await
+            .expect("shard file should be written");
+
+        temp_env::async_with_vars([(ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("1024"))], async {
+            let over_cap = open_disk_reader(&disk, "test-bucket", "obj/part.1", 0, payload.len(), true, None)
+                .await
+                .expect("over-cap read should open");
+            assert!(
+                matches!(over_cap, ShardReader::Stream(_)),
+                "read longer than the mmap cap must take the streaming path"
+            );
+            let mut over_cap = over_cap;
+            let mut streamed = Vec::new();
+            over_cap
+                .read_to_end(&mut streamed)
+                .await
+                .expect("streaming fallback should read the range");
+            assert_eq!(streamed, payload, "streaming fallback must return the same bytes");
+
+            let under_cap = open_disk_reader(&disk, "test-bucket", "obj/part.1", 0, 512, true, None)
+                .await
+                .expect("under-cap read should open");
+            assert!(
+                matches!(under_cap, ShardReader::InMemory(_)),
+                "read within the mmap cap keeps the mmap-copy fast path"
+            );
+
+            let at_cap = open_disk_reader(&disk, "test-bucket", "obj/part.1", 0, 1024, true, None)
+                .await
+                .expect("at-cap read should open");
+            assert!(
+                matches!(at_cap, ShardReader::InMemory(_)),
+                "read of exactly the mmap cap keeps the mmap-copy fast path"
+            );
+        })
+        .await;
+
+        // Cap of 0 disables mmap-copy for every non-empty read.
+        temp_env::async_with_vars([(ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("0"))], async {
+            let disabled = open_disk_reader(&disk, "test-bucket", "obj/part.1", 0, 512, true, None)
+                .await
+                .expect("read with cap 0 should open");
+            assert!(
+                matches!(disabled, ShardReader::Stream(_)),
+                "cap 0 must route every non-empty read to the streaming path"
+            );
+        })
+        .await;
+    }
+
    #[tokio::test]
    async fn test_create_bitrot_reader_with_inline_data() {
        let test_data = b"hello world test data";
@@ -1020,18 +1020,14 @@ impl EndpointServerPools {

    #[instrument]
    pub fn hosts_sorted(&self) -> Vec<Option<XHost>> {
-        let (mut peers, local) = self.peers();
+        let peers = self.peer_grid_hosts_sorted();

        let mut ret = vec![None; peers.len()];
-
-        peers.sort();
-
-        for (i, peer) in peers.iter().enumerate() {
-            if &local == peer {
+        for (i, peer) in peers.into_iter().enumerate() {
+            let Some((peer_host_port, _)) = peer else {
                continue;
-            }
-
-            let host = match XHost::try_from(peer.clone()) {
+            };
+            let host = match XHost::try_from(peer_host_port) {
                Ok(res) => res,
                Err(err) => {
                    warn!("Xhost parse failed {:?}", err);
@@ -1044,6 +1040,61 @@ impl EndpointServerPools {

        ret
    }
+
+    pub fn peer_host_ports_sorted(&self) -> Vec<Option<String>> {
+        let (mut peers, local) = self.peers();
+        let mut ret = vec![None; peers.len()];
+
+        peers.sort();
+
+        for (i, peer) in peers.into_iter().enumerate() {
+            if local == peer {
+                continue;
+            }
+            ret[i] = Some(peer);
+        }
+
+        ret
+    }
+
+    pub fn peer_grid_hosts_sorted(&self) -> Vec<Option<(String, String)>> {
+        self.peer_grid_host_slots_sorted()
+            .into_iter()
+            .map(|(peer, grid_host, is_local)| {
+                if is_local {
+                    None
+                } else {
+                    grid_host.map(|grid_host| (peer, grid_host))
+                }
+            })
+            .collect()
+    }
+
+    pub fn peer_grid_host_slots_sorted(&self) -> Vec<(String, Option<String>, bool)> {
+        let (mut peers, local) = self.peers();
+        let mut grid_hosts = HashMap::with_capacity(peers.len());
+
+        for ep in self.0.iter() {
+            for endpoint in ep.endpoints.0.iter() {
+                if endpoint.get_type() != EndpointType::Url || endpoint.is_local {
+                    continue;
+                }
+                grid_hosts.entry(endpoint.host_port()).or_insert_with(|| endpoint.grid_host());
+            }
+        }
+
+        peers.sort();
+
+        peers
+            .into_iter()
+            .map(|peer| {
+                let is_local = local == peer;
+                let grid_host = if is_local { None } else { grid_hosts.get(&peer).cloned() };
+                (peer, grid_host, is_local)
+            })
+            .collect()
+    }
+
    pub fn peers(&self) -> (Vec<String>, String) {
        let mut local = None;
        let mut set = HashSet::new();
@@ -1087,6 +1138,25 @@ impl EndpointServerPools {

        None
    }
+
+    pub fn find_grid_host_from_peer_host_port(&self, peer_host_port: &str) -> Option<String> {
+        for ep in self.0.iter() {
+            for endpoint in ep.endpoints.0.iter() {
+                if endpoint.is_local {
+                    continue;
+                }
+                if endpoint.host_port() == peer_host_port {
+                    return Some(endpoint.grid_host());
+                }
+            }
+        }
+
+        if let Ok(host) = XHost::try_from(peer_host_port.to_owned()) {
+            return self.find_grid_hosts_from_peer(&host);
+        }
+
+        None
+    }
 }

 fn validate_local_physical_disk_independence(pools: &[Endpoints]) -> Result<()> {
@@ -1407,6 +1477,43 @@ mod test {
        assert!(!eps[2].is_local, "a genuinely remote host must stay remote");
    }

+    #[test]
+    fn peer_host_ports_sorted_preserves_raw_remote_hosts() {
+        let mut local =
+            Endpoint::try_from("http://rustfs-1.storage.swarm.private:9000/data").expect("local endpoint should parse");
+        local.is_local = true;
+        let mut remote =
+            Endpoint::try_from("http://rustfs-4.storage.swarm.private:9000/data").expect("remote endpoint should parse");
+        remote.is_local = false;
+        let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(vec![local, remote]),
+            cmd_line: String::new(),
+            platform: String::new(),
+        }]);
+
+        let peers = endpoint_pools.peer_host_ports_sorted();
+
+        assert_eq!(
+            peers,
+            vec![None, Some("rustfs-4.storage.swarm.private:9000".to_string())],
+            "membership enumeration must keep raw topology host:port instead of requiring DNS resolution"
+        );
+        assert_eq!(
+            endpoint_pools
+                .find_grid_host_from_peer_host_port("rustfs-4.storage.swarm.private:9000")
+                .as_deref(),
+            Some("http://rustfs-4.storage.swarm.private:9000"),
+            "raw host:port should map directly to the configured grid host"
+        );
+        let peer_grid_hosts = endpoint_pools.peer_grid_hosts_sorted();
+        let remote = peer_grid_hosts[1].as_ref().expect("remote peer should be resolved");
+        assert_eq!(remote.0, "rustfs-4.storage.swarm.private:9000");
+        assert_eq!(remote.1, "http://rustfs-4.storage.swarm.private:9000");
+    }
+
    #[tokio::test]
    async fn orchestrated_policy_defers_dns_ip_same_path_check() {
        // Two remote endpoints on the same address, same path, different ports
@@ -95,16 +95,18 @@ pub fn get_global_notification_sys() -> Option<Arc<NotificationSys>> {
 pub struct NotificationSys {
    pub peer_clients: Vec<Option<PeerRestClient>>,
    pub all_peer_clients: Vec<Option<PeerRestClient>>,
+    peer_topology_hosts: Vec<String>,
    peer_admin_caches: Vec<Mutex<PeerAdminCache>>,
 }

 impl NotificationSys {
    pub async fn new(eps: EndpointServerPools) -> Self {
-        let (peer_clients, all_peer_clients) = PeerRestClient::new_clients(eps).await;
+        let (peer_clients, all_peer_clients, peer_topology_hosts) = PeerRestClient::new_clients_with_topology(eps).await;
        let peer_admin_caches = (0..peer_clients.len()).map(|_| Mutex::new(PeerAdminCache::new())).collect();
        Self {
            peer_clients,
            all_peer_clients,
+            peer_topology_hosts,
            peer_admin_caches,
        }
    }
@@ -379,22 +381,19 @@ impl NotificationSys {
        let peer_timeout = Duration::from_secs(5);

        for (idx, client) in self.peer_clients.iter().enumerate() {
-            let endpoints = endpoints.clone();
-            let cache = self.peer_admin_caches.get(idx);
+            let host = self
+                .peer_topology_hosts
+                .get(idx)
+                .cloned()
+                .or_else(|| client.as_ref().map(|client| client.host.to_string()))
+                .unwrap_or_default();
            futures.push(async move {
-                // `peer_clients` comes from `new_clients`, which only ever pushes
-                // `Some(client)` (local hosts are excluded, not slotted as
-                // `None`), so this branch is unreachable in practice. Kept as a
-                // defensive fallback: report an explicit `unknown` state rather
-                // than a blank `default()` entry, so it can never contribute a
-                // hollow row to `servers[]` (rustfs/backlog#1049 P3).
                let Some(client) = client else {
-                    return ServerProperties {
-                        state: ItemState::Unknown.to_string().to_owned(),
-                        ..Default::default()
+                    return PeerServerInfoProbe {
+                        host,
+                        result: Err(PeerServerInfoProbeFailure::NoClient),
                    };
                };
-                let host = client.host.to_string();

                // First attempt. A single evicted or half-open internode channel
                // is enough to fail one probe and, before retrying, would drop
@@ -403,8 +402,7 @@ impl NotificationSys {
                // before falling back (rustfs/backlog#1049, P1-B).
                match timeout(peer_timeout, client.server_info()).await {
                    Ok(Ok(info)) => {
-                        update_server_info_cache(cache, &host, &info);
-                        return info;
+                        return PeerServerInfoProbe { host, result: Ok(info) };
                    }
                    Ok(Err(err)) => debug!("peer {host} server_info failed (attempt 1/2): {err}"),
                    Err(_) => debug!("peer {host} server_info timed out (attempt 1/2) after {peer_timeout:?}"),
@@ -420,26 +418,29 @@ impl NotificationSys {

                // Second and final attempt on the fresh channel.
                match timeout(peer_timeout, client.server_info()).await {
-                    Ok(Ok(info)) => {
-                        update_server_info_cache(cache, &host, &info);
-                        info
-                    }
+                    Ok(Ok(info)) => PeerServerInfoProbe { host, result: Ok(info) },
                    Ok(Err(err)) => {
                        warn!("peer {host} server_info failed after retry: {err}");
                        let health = peer_disk_health(&host).await;
-                        handle_server_info_failure(cache, &host, &endpoints, health.as_ref())
+                        PeerServerInfoProbe {
+                            host,
+                            result: Err(PeerServerInfoProbeFailure::Rpc { health }),
+                        }
                    }
                    Err(_) => {
                        warn!("peer {host} server_info timed out after retry ({peer_timeout:?})");
                        client.evict_connection().await;
                        let health = peer_disk_health(&host).await;
-                        handle_server_info_failure(cache, &host, &endpoints, health.as_ref())
+                        PeerServerInfoProbe {
+                            host,
+                            result: Err(PeerServerInfoProbeFailure::Rpc { health }),
+                        }
                    }
                }
            });
        }

-        join_all(futures).await
+        publish_server_info_probe_round(&self.peer_admin_caches, &endpoints, join_all(futures).await)
    }

    pub async fn load_user(&self, access_key: &str, temp: bool) -> Vec<NotificationPeerErr> {
@@ -1263,6 +1264,16 @@ struct PeerDiskHealth {
    disks: Vec<rustfs_madmin::Disk>,
 }

+struct PeerServerInfoProbe {
+    host: String,
+    result: std::result::Result<ServerProperties, PeerServerInfoProbeFailure>,
+}
+
+enum PeerServerInfoProbeFailure {
+    Rpc { health: Option<PeerDiskHealth> },
+    NoClient,
+}
+
 /// Consult the local disk-health state for `host` without issuing any RPC.
 ///
 /// On the aggregating node a peer's drives are remote-disk handles whose
@@ -1424,6 +1435,30 @@ fn handle_server_info_failure(
    unknown_server_properties(host, endpoints)
 }

+fn publish_server_info_probe_round(
+    caches: &[Mutex<PeerAdminCache>],
+    endpoints: &EndpointServerPools,
+    probes: Vec<PeerServerInfoProbe>,
+) -> Vec<ServerProperties> {
+    probes
+        .into_iter()
+        .enumerate()
+        .map(|(idx, probe)| {
+            let cache = caches.get(idx);
+            match probe.result {
+                Ok(info) => {
+                    update_server_info_cache(cache, &probe.host, &info);
+                    info
+                }
+                Err(PeerServerInfoProbeFailure::Rpc { health }) => {
+                    handle_server_info_failure(cache, &probe.host, endpoints, health.as_ref())
+                }
+                Err(PeerServerInfoProbeFailure::NoClient) => unknown_server_properties(&probe.host, endpoints),
+            }
+        })
+        .collect()
+}
+
 fn update_server_info_cache(cache: Option<&Mutex<PeerAdminCache>>, host: &str, info: &ServerProperties) {
    let Some(cache) = cache else {
        return;
@@ -1634,6 +1669,7 @@ mod tests {
                "127.0.0.1:9000".to_string().try_into().expect("peer host should parse"),
                "http://127.0.0.1:9000".to_string(),
            ))],
+            peer_topology_hosts: Vec::new(),
            peer_admin_caches: Vec::new(),
        };

@@ -1677,6 +1713,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![None],
            all_peer_clients: Vec::new(),
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };

@@ -1696,6 +1733,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![None],
            all_peer_clients: vec![None, None],
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };

@@ -1712,6 +1750,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: Vec::new(),
            all_peer_clients: Vec::new(),
+            peer_topology_hosts: Vec::new(),
            peer_admin_caches: Vec::new(),
        };

@@ -1732,6 +1771,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![Some(client)],
            all_peer_clients: vec![None],
+            peer_topology_hosts: vec!["127.0.0.1:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };

@@ -1743,6 +1783,51 @@ mod tests {
        assert!(err.to_string().contains("peer topology is incomplete"));
    }

+    #[tokio::test]
+    async fn server_info_no_client_slot_uses_topology_host_without_counting_rpc_failure() {
+        let sys = NotificationSys {
+            peer_clients: vec![None],
+            all_peer_clients: vec![None, None],
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
+            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
+        };
+
+        let servers = sys.server_info().await;
+
+        assert_eq!(servers.len(), 1);
+        assert_eq!(servers[0].endpoint, "node-a:9000");
+        assert_eq!(servers[0].state, ItemState::Unknown.to_string());
+        let cache = sys.peer_admin_caches[0].lock().expect("cache mutex should not be poisoned");
+        assert_eq!(cache.server_failures, 0, "construction-only missing slots are not failed RPC attempts");
+        assert!(cache.last_server_info.is_none());
+    }
+
+    #[test]
+    fn server_info_failure_cache_stays_aligned_with_topology_slot() {
+        let cache_a = Mutex::new(PeerAdminCache {
+            last_server_info: Some(build_props("cached-a")),
+            last_server_success: Some(SystemTime::now()),
+            server_failures: 1,
+            storage_failures: 0,
+            last_storage_info: None,
+        });
+        let cache_b = Mutex::new(PeerAdminCache {
+            last_server_info: Some(build_props("cached-b")),
+            last_server_success: Some(SystemTime::now()),
+            server_failures: 1,
+            storage_failures: 0,
+            last_storage_info: None,
+        });
+        let caches = [cache_a, cache_b];
+        let endpoints = EndpointServerPools::from(Vec::new());
+
+        let rendered = handle_server_info_failure(Some(&caches[1]), "node-b:9000", &endpoints, None);
+
+        assert_eq!(rendered.endpoint, "cached-b");
+        assert_eq!(caches[0].lock().expect("cache mutex should not be poisoned").server_failures, 1);
+        assert_eq!(caches[1].lock().expect("cache mutex should not be poisoned").server_failures, 2);
+    }
+
    #[tokio::test]
    async fn scanner_activity_probe_times_out() {
        let err = scanner_activity_with_timeout(
@@ -1762,6 +1847,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![None],
            all_peer_clients: Vec::new(),
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };

@@ -1781,6 +1867,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![None],
            all_peer_clients: Vec::new(),
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };

@@ -1796,6 +1883,7 @@ mod tests {
        let sys = NotificationSys {
            peer_clients: vec![None],
            all_peer_clients: Vec::new(),
+            peer_topology_hosts: vec!["node-a:9000".to_string()],
            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
        };
        let mutation_id = Uuid::from_u128(1);
@@ -1996,6 +2084,60 @@ mod tests {
        assert!(!cached_snapshot_is_fresh(Some(stale)), "an old success is stale");
    }

+    #[test]
+    fn server_info_probe_round_commits_failures_only_when_published() {
+        let caches = vec![Mutex::new(PeerAdminCache::new())];
+        let endpoints = EndpointServerPools::default();
+        let probes = vec![PeerServerInfoProbe {
+            host: "peer-1".to_string(),
+            result: Err(PeerServerInfoProbeFailure::Rpc { health: None }),
+        }];
+
+        assert_eq!(
+            caches[0]
+                .lock()
+                .expect("peer cache should lock before publish")
+                .server_failures,
+            0
+        );
+
+        let replies = publish_server_info_probe_round(&caches, &endpoints, probes);
+
+        assert_eq!(replies.len(), 1);
+        assert_eq!(replies[0].endpoint, "peer-1");
+        assert_eq!(replies[0].state, ItemState::Unknown.to_string());
+        assert_eq!(
+            caches[0]
+                .lock()
+                .expect("peer cache should lock after publish")
+                .server_failures,
+            1
+        );
+    }
+
+    #[test]
+    fn server_info_probe_round_does_not_count_no_client_slots_as_rpc_failures() {
+        let caches = vec![Mutex::new(PeerAdminCache::new())];
+        let endpoints = EndpointServerPools::default();
+        let probes = vec![PeerServerInfoProbe {
+            host: "node-a:9000".to_string(),
+            result: Err(PeerServerInfoProbeFailure::NoClient),
+        }];
+
+        let replies = publish_server_info_probe_round(&caches, &endpoints, probes);
+
+        assert_eq!(replies.len(), 1);
+        assert_eq!(replies[0].endpoint, "node-a:9000");
+        assert_eq!(replies[0].state, ItemState::Unknown.to_string());
+        assert_eq!(
+            caches[0]
+                .lock()
+                .expect("peer cache should lock after no-client publish")
+                .server_failures,
+            0
+        );
+    }
+
    #[test]
    fn endpoint_host_matches_direct_and_canonicalized() {
        // Direct match (IP deployment): peer host already equals host_port.
@@ -172,21 +172,31 @@ pub fn oidc_plugin_authn_metrics_snapshot() -> OidcPluginAuthnMetricsSnapshot {
    OIDC_PLUGIN_AUTHN_METRICS.snapshot()
 }

+/// Header names whose values may carry OIDC secrets (client credentials, cookies,
+/// bearer tokens). Their values are never emitted to logs, only their byte length.
+const SENSITIVE_HEADER_NAMES: [&str; 4] = ["authorization", "proxy-authorization", "cookie", "set-cookie"];
+
+fn is_sensitive_header(name: &str) -> bool {
+    SENSITIVE_HEADER_NAMES
+        .iter()
+        .any(|candidate| name.eq_ignore_ascii_case(candidate))
+}
+
 fn format_http_headers(headers: &http::HeaderMap) -> String {
    headers
        .iter()
        .map(|(name, value)| {
-            let value = value.to_str().unwrap_or("<non-utf8>");
-            format!("{}={}", name.as_str(), value)
+            if is_sensitive_header(name.as_str()) {
+                format!("{}=<redacted len={}>", name.as_str(), value.as_bytes().len())
+            } else {
+                let value = value.to_str().unwrap_or("<non-utf8>");
+                format!("{}={}", name.as_str(), value)
+            }
        })
        .collect::<Vec<_>>()
        .join("; ")
 }

-fn format_http_body(body: &[u8]) -> String {
-    String::from_utf8_lossy(body).into_owned()
-}
-
 #[derive(Debug, Default)]
 struct TokenResponseBodyShape {
    json_object: bool,
@@ -326,7 +336,6 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
            let uri = parts.uri.to_string();
            if tracing::enabled!(tracing::Level::DEBUG) {
                let request_headers = format_http_headers(&parts.headers);
-                let request_body = format_http_body(&body);
                debug!(
                    event = EVENT_OIDC_HTTP,
                    component = LOG_COMPONENT_IAM,
@@ -336,7 +345,6 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
                    uri = %uri,
                    request_headers = %request_headers,
                    request_body_len = body.len(),
-                    request_body = %request_body,
                    "oidc outbound http"
                );
            }
@@ -387,7 +395,6 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
            })?;
            if tracing::enabled!(tracing::Level::DEBUG) {
                let response_headers = format_http_headers(&headers);
-                let response_body = format_http_body(&body_bytes);
                debug!(
                    event = EVENT_OIDC_HTTP,
                    component = LOG_COMPONENT_IAM,
@@ -400,7 +407,6 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
                    elapsed_ms,
                    response_headers = %response_headers,
                    response_body_len = body_bytes.len(),
-                    response_body = %response_body,
                    "oidc outbound http"
                );
            }
@@ -814,7 +820,6 @@ impl OidcSys {
                }
                RequestTokenError::Parse(parse_err, body) => {
                    let shape = inspect_token_response_body(body);
-                    let response_body = format_http_body(body);
                    error!(
                        event = EVENT_OIDC_DIAGNOSTICS,
                        component = LOG_COMPONENT_IAM,
@@ -837,12 +842,11 @@ impl OidcSys {
                        response_has_error = shape.has_error,
                        response_has_error_description = shape.has_error_description,
                        response_looks_like_html = shape.looks_like_html,
-                        response_body = %response_body,
                        error = %e,
                        "oidc token exchange failed"
                    );
                    format!(
-                        "token exchange failed: {e}: stage=token_response_parse_failed, provider_id={}, config_url={}, issuer={}, token_endpoint={}, redirect_uri={}, client_id={}, parse_error_path={}, response_body_len={}, response_json_keys={}, response_has_id_token={}, response_has_error={}, response_looks_like_html={}, response_body={}",
+                        "token exchange failed: {e}: stage=token_response_parse_failed, provider_id={}, config_url={}, issuer={}, token_endpoint={}, redirect_uri={}, client_id={}, parse_error_path={}, response_body_len={}, response_json_keys={}, response_has_id_token={}, response_has_error={}, response_looks_like_html={}",
                        session.provider_id,
                        config.config_url,
                        issuer,
@@ -854,8 +858,7 @@ impl OidcSys {
                        shape.json_keys,
                        shape.has_id_token,
                        shape.has_error,
-                        shape.looks_like_html,
-                        response_body
+                        shape.looks_like_html
                    )
                }
                RequestTokenError::Other(message) => {
@@ -1849,6 +1852,39 @@ mod tests {
        assert_eq!(extract_string_claim(&claims, "missing"), "");
    }

+    #[test]
+    fn format_http_headers_redacts_sensitive_values() {
+        let mut headers = http::HeaderMap::new();
+        headers.insert(http::header::AUTHORIZATION, "Basic Y2xpZW50OnNlY3JldA==".parse().unwrap());
+        headers.insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap());
+        headers.insert(http::header::COOKIE, "session=super-secret".parse().unwrap());
+
+        let rendered = format_http_headers(&headers);
+
+        // Sensitive header values never appear; only their length is emitted.
+        assert!(!rendered.contains("Y2xpZW50OnNlY3JldA=="), "authorization value leaked: {rendered}");
+        assert!(!rendered.contains("super-secret"), "cookie value leaked: {rendered}");
+        assert!(
+            rendered.contains("authorization=<redacted len="),
+            "expected redacted authorization: {rendered}"
+        );
+        assert!(rendered.contains("cookie=<redacted len="), "expected redacted cookie: {rendered}");
+        // Non-sensitive header values are preserved for diagnostics.
+        assert!(
+            rendered.contains("content-type=application/json"),
+            "content-type should be visible: {rendered}"
+        );
+    }
+
+    #[test]
+    fn is_sensitive_header_is_case_insensitive() {
+        assert!(is_sensitive_header("Authorization"));
+        assert!(is_sensitive_header("PROXY-AUTHORIZATION"));
+        assert!(is_sensitive_header("Set-Cookie"));
+        assert!(!is_sensitive_header("content-type"));
+        assert!(!is_sensitive_header("x-request-id"));
+    }
+
    #[test]
    fn test_extract_groups_claim_array() {
        let mut claims = HashMap::new();
@@ -1156,10 +1156,50 @@ pub struct SRStateEditReq {
 pub struct ResyncBucketStatus {
    #[serde(default)]
    pub bucket: String,
+    #[serde(rename = "targetArn", default, skip_serializing_if = "String::is_empty")]
+    pub target_arn: String,
    #[serde(default)]
    pub status: String,
    #[serde(rename = "errorDetail", skip_serializing_if = "String::is_empty", default)]
    pub err_detail: String,
+    #[serde(
+        rename = "createdAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub created_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "startedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub started_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "updatedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub updated_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "completedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub completed_at: Option<OffsetDateTime>,
+    #[serde(default, skip_serializing_if = "is_zero_u64")]
+    pub generation: u64,
+    #[serde(rename = "replicatedObjects", default, skip_serializing_if = "is_zero_u64")]
+    pub replicated_objects: u64,
+    #[serde(rename = "replicatedBytes", default, skip_serializing_if = "is_zero_u64")]
+    pub replicated_bytes: u64,
+    #[serde(rename = "failedObjects", default, skip_serializing_if = "is_zero_u64")]
+    pub failed_objects: u64,
+    #[serde(rename = "failedBytes", default, skip_serializing_if = "is_zero_u64")]
+    pub failed_bytes: u64,
 }

 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -1170,10 +1210,74 @@ pub struct SRResyncOpStatus {
    pub resync_id: String,
    #[serde(default)]
    pub status: String,
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub state: String,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub buckets: Vec<ResyncBucketStatus>,
    #[serde(rename = "errorDetail", skip_serializing_if = "String::is_empty", default)]
    pub err_detail: String,
+    #[serde(
+        rename = "createdAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub created_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "startedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub started_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "updatedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub updated_at: Option<OffsetDateTime>,
+    #[serde(
+        rename = "completedAt",
+        default,
+        with = "time::serde::rfc3339::option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub completed_at: Option<OffsetDateTime>,
+    #[serde(default, skip_serializing_if = "is_zero_u64")]
+    pub generation: u64,
+    #[serde(rename = "totalBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub total_buckets: u64,
+    #[serde(rename = "pendingBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub pending_buckets: u64,
+    #[serde(rename = "runningBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub running_buckets: u64,
+    #[serde(rename = "completedBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub completed_buckets: u64,
+    #[serde(rename = "failedBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub failed_buckets: u64,
+    #[serde(rename = "canceledBuckets", default, skip_serializing_if = "is_zero_u64")]
+    pub canceled_buckets: u64,
+    #[serde(rename = "replicatedObjects", default, skip_serializing_if = "is_zero_u64")]
+    pub replicated_objects: u64,
+    #[serde(rename = "replicatedBytes", default, skip_serializing_if = "is_zero_u64")]
+    pub replicated_bytes: u64,
+    #[serde(rename = "failedObjects", default, skip_serializing_if = "is_zero_u64")]
+    pub failed_objects: u64,
+    #[serde(rename = "failedBytes", default, skip_serializing_if = "is_zero_u64")]
+    pub failed_bytes: u64,
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub truncated: bool,
+    #[serde(rename = "nextContinuationToken", default, skip_serializing_if = "String::is_empty")]
+    pub next_continuation_token: String,
+}
+
+fn is_zero_u64(value: &u64) -> bool {
+    *value == 0
+}
+
+fn is_false(value: &bool) -> bool {
+    !*value
 }

 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -1202,7 +1306,7 @@ pub struct SiteNetPerfResult {

 #[cfg(test)]
 mod tests {
-    use super::{PeerInfo, PeerSite};
+    use super::{PeerInfo, PeerSite, SRResyncOpStatus};
    use serde_json::{Value, json};

    const TEST_CA_CERT: &str = "-----BEGIN CERTIFICATE-----\ntest-ca\n-----END CERTIFICATE-----";
@@ -1302,4 +1406,94 @@ mod tests {
        assert!(peer_debug.contains("skip_tls_verify: false"));
        assert!(peer_debug.contains("has_custom_ca: true"));
    }
+
+    #[test]
+    fn resync_status_legacy_json_defaults_new_lifecycle_fields() {
+        let legacy_json = json!({
+            "op": "start",
+            "id": "resync-1",
+            "status": "success",
+            "buckets": [{
+                "bucket": "photos",
+                "status": "success"
+            }]
+        });
+
+        let status: SRResyncOpStatus =
+            serde_json::from_value(legacy_json.clone()).expect("legacy resync status should deserialize");
+
+        assert_eq!(status.generation, 0);
+        assert!(status.state.is_empty());
+        assert!(status.created_at.is_none());
+        assert_eq!(status.total_buckets, 0);
+        assert_eq!(status.replicated_objects, 0);
+        assert!(!status.truncated);
+        assert!(status.next_continuation_token.is_empty());
+        assert!(status.buckets[0].created_at.is_none());
+        assert!(status.buckets[0].target_arn.is_empty());
+        assert_eq!(status.buckets[0].generation, 0);
+        assert_eq!(status.buckets[0].replicated_bytes, 0);
+        assert_eq!(serde_json::to_value(status).expect("legacy resync status should serialize"), legacy_json);
+    }
+
+    #[test]
+    fn resync_status_lifecycle_fields_round_trip_with_exact_json_names() {
+        let status_json = json!({
+            "op": "status",
+            "id": "resync-2",
+            "status": "success",
+            "state": "running",
+            "createdAt": "2026-07-22T01:00:00Z",
+            "startedAt": "2026-07-22T01:00:01Z",
+            "updatedAt": "2026-07-22T01:01:00Z",
+            "completedAt": "2026-07-22T01:02:00Z",
+            "generation": 7,
+            "totalBuckets": 6,
+            "pendingBuckets": 1,
+            "runningBuckets": 1,
+            "completedBuckets": 1,
+            "failedBuckets": 1,
+            "canceledBuckets": 2,
+            "replicatedObjects": 12,
+            "replicatedBytes": 4096,
+            "failedObjects": 3,
+            "failedBytes": 512,
+            "truncated": true,
+            "nextContinuationToken": "bucket-page-2",
+            "buckets": [{
+                "bucket": "photos",
+                "targetArn": "arn:rustfs:replication::peer-a:photos",
+                "status": "failed",
+                "errorDetail": "target unavailable",
+                "createdAt": "2026-07-22T01:00:00Z",
+                "startedAt": "2026-07-22T01:00:01Z",
+                "updatedAt": "2026-07-22T01:01:00Z",
+                "completedAt": "2026-07-22T01:02:00Z",
+                "generation": 7,
+                "replicatedObjects": 12,
+                "replicatedBytes": 4096,
+                "failedObjects": 3,
+                "failedBytes": 512
+            }]
+        });
+
+        let status: SRResyncOpStatus =
+            serde_json::from_value(status_json.clone()).expect("expanded resync status should deserialize");
+
+        assert_eq!(status.generation, 7);
+        assert_eq!(status.state, "running");
+        assert_eq!(status.total_buckets, 6);
+        assert_eq!(status.completed_buckets, 1);
+        assert_eq!(status.replicated_bytes, 4096);
+        assert!(status.truncated);
+        assert_eq!(status.next_continuation_token, "bucket-page-2");
+        assert_eq!(status.buckets[0].generation, 7);
+        assert_eq!(status.buckets[0].target_arn, "arn:rustfs:replication::peer-a:photos");
+        assert_eq!(status.buckets[0].failed_objects, 3);
+        assert!(status.buckets[0].completed_at.is_some());
+        assert_eq!(
+            serde_json::to_value(status).expect("expanded resync status should serialize"),
+            status_json
+        );
+    }
 }
@@ -154,7 +154,7 @@ impl TargetReplicationResyncStatus {
    }

    fn marshal_wire_msg(&self, wr: &mut Vec<u8>) -> Result<()> {
-        rmp::encode::write_map_len(wr, 11)?;
+        rmp::encode::write_map_len(wr, 12)?;
        rmp::encode::write_str(wr, "st")?;
        write_msgp_time(wr, wire_time_or_default(self.start_time))?;
        rmp::encode::write_str(wr, "lst")?;
@@ -177,6 +177,8 @@ impl TargetReplicationResyncStatus {
        rmp::encode::write_str(wr, &self.bucket)?;
        rmp::encode::write_str(wr, "obj")?;
        rmp::encode::write_str(wr, &self.object)?;
+        rmp::encode::write_str(wr, "err")?;
+        rmp::encode::write_str(wr, self.error.as_deref().unwrap_or_default())?;
        Ok(())
    }

@@ -202,6 +204,10 @@ impl TargetReplicationResyncStatus {
                "rrc" => out.replicated_count = rmp::decode::read_int(rd)?,
                "bkt" => out.bucket = read_msgp_str(rd)?,
                "obj" => out.object = read_msgp_str(rd)?,
+                "err" => {
+                    let error = read_msgp_str(rd)?;
+                    out.error = (!error.is_empty()).then_some(error);
+                }
                _ => skip_msgp_value(rd)?,
            }
        }
@@ -623,6 +629,7 @@ mod tests {
                bucket: "bucket-a".to_string(),
                object: "object-a".to_string(),
                replicated_count: 7,
+                error: Some("durable failure".to_string()),
                ..Default::default()
            },
        );
@@ -634,6 +641,7 @@ mod tests {
        assert_eq!(got.targets_map["arn:replication:a"].resync_id, "rid-1");
        assert_eq!(got.targets_map["arn:replication:a"].resync_status, ResyncStatusType::ResyncStarted);
        assert_eq!(got.targets_map["arn:replication:a"].replicated_count, 7);
+        assert_eq!(got.targets_map["arn:replication:a"].error.as_deref(), Some("durable failure"));
    }

    #[test]
@@ -22,6 +22,7 @@ pub enum CapabilityState {
    Supported,
    Unsupported,
    Disabled,
+    #[serde(other)]
    #[default]
    Unknown,
 }
@@ -133,4 +134,12 @@ mod tests {
        assert_eq!(decoded.1.reason.as_deref(), Some("target does not expose profiler"));
        assert!(!decoded.1.state.is_supported());
    }
+
+    #[test]
+    fn future_capability_state_deserializes_as_unknown() {
+        let status: CapabilityStatus = serde_json::from_str(r#"{"state":"future_state"}"#)
+            .expect("future capability states should preserve conservative compatibility");
+
+        assert_eq!(status.state, CapabilityState::Unknown);
+    }
 }
@@ -23,7 +23,7 @@ use rustfs_config::{
    NATS_TLS_CLIENT_KEY, NATS_TOKEN, NATS_USERNAME, PULSAR_AUTH_TOKEN, PULSAR_PASSWORD, PULSAR_QUEUE_DIR, PULSAR_TLS_CA,
    PULSAR_TOPIC, PULSAR_USERNAME,
 };
-use rustfs_utils::egress::validate_outbound_url;
+use rustfs_utils::egress::OutboundPolicy;
 use std::collections::HashSet;
 use std::path::Path;
 use std::str::FromStr;
@@ -215,16 +215,20 @@ pub(super) fn validate_pulsar_broker_config(broker: &str, config: &KVS, default_
 }

 pub(super) fn parse_url(value: &str, field_label: &str) -> Result<Url, TargetError> {
-    Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e} (value: '{value}')")))
+    Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e}")))
 }

 pub(super) fn validate_outbound_http_url(value: &Url, field_label: &str) -> Result<(), TargetError> {
-    validate_outbound_url(value).map_err(|e| TargetError::Configuration(format!("{field_label} is not allowed: {e}")))
+    let policy =
+        OutboundPolicy::from_env_cached().map_err(|err| TargetError::Configuration(format!("invalid outbound policy: {err}")))?;
+    policy
+        .validate_url(value)
+        .map_err(|e| TargetError::Configuration(format!("{field_label} is not allowed: {e}")))
 }

 #[cfg(test)]
 mod tests {
-    use super::{parse_jetstream_enable, validate_nats_server_config, validate_pulsar_broker_config};
+    use super::{parse_jetstream_enable, parse_url, validate_nats_server_config, validate_pulsar_broker_config};
    use async_nats::ServerAddr;
    use rustfs_config::server_config::KVS;
    use rustfs_config::{
@@ -238,6 +242,12 @@ mod tests {
        ServerAddr::from_str("nats://127.0.0.1:4222").expect("valid nats address")
    }

+    #[test]
+    fn parse_url_error_does_not_echo_the_configured_value() {
+        let err = parse_url("not a URL containing secret-token", "endpoint URL").expect_err("invalid URL should fail");
+        assert!(!err.to_string().contains("secret-token"));
+    }
+
    // Absolute on Linux, macOS, and Windows. temp_dir needs no filesystem to exist for a
    // validation-only test, and Path::is_absolute stays true across platforms.
    fn nats_queue_dir() -> String {
@@ -84,7 +84,23 @@ fn redact_target_field_value(field_name: &str, value: &str) -> String {
        return crate::target::mysql::redact_mysql_dsn(value);
    }
    if is_sensitive_target_field(field_name) {
-        return "***redacted***".to_string();
+        return crate::target::REDACTED_SECRET.to_string();
+    }
+    if field_name.eq_ignore_ascii_case(rustfs_config::WEBHOOK_ENDPOINT)
+        || field_name.eq_ignore_ascii_case(rustfs_config::AMQP_URL)
+    {
+        return url::Url::parse(value)
+            .ok()
+            .and_then(|endpoint| {
+                let host = match endpoint.host()? {
+                    url::Host::Domain(host) => host.to_string(),
+                    url::Host::Ipv4(host) => host.to_string(),
+                    url::Host::Ipv6(host) => format!("[{host}]"),
+                };
+                let port = endpoint.port().map(|port| format!(":{port}")).unwrap_or_default();
+                Some(format!("{}://{host}{port}", endpoint.scheme()))
+            })
+            .unwrap_or_else(|| crate::target::REDACTED_SECRET.to_string());
    }
    value.to_string()
 }
@@ -632,6 +648,24 @@ mod tests {
        assert_eq!(redact_target_field_value("queue_limit", "1000"), "1000");
    }

+    #[test]
+    fn redact_target_field_value_strips_endpoint_path_and_query() {
+        assert_eq!(
+            redact_target_field_value("endpoint", "https://example.com/private/hook?token=secret"),
+            "https://example.com"
+        );
+        assert_eq!(redact_target_field_value("endpoint", "not a URL with secret"), "***redacted***");
+    }
+
+    #[test]
+    fn redact_target_field_value_strips_url_credentials() {
+        assert_eq!(
+            redact_target_field_value("url", "amqps://user:secret@broker.example/vhost"),
+            "amqps://broker.example"
+        );
+        assert_eq!(redact_target_field_value("url", "not a URL with secret"), "***redacted***");
+    }
+
    #[test]
    fn redact_dsn_string_partial_redaction() {
        let dsn = "rustfs:secret123@tcp(mysql.example.com:3306)/rustfs_events";
@@ -671,7 +705,7 @@ mod tests {
        assert_eq!(
            redacted,
            vec![
-                ("endpoint".to_string(), "https://example.com/hook".to_string()),
+                ("endpoint".to_string(), "https://example.com".to_string()),
                ("password".to_string(), "***redacted***".to_string()),
                ("client_key".to_string(), "***redacted***".to_string()),
                ("auth_token".to_string(), "***redacted***".to_string()),
@@ -32,7 +32,7 @@ use async_trait::async_trait;
 use parking_lot::Mutex;
 use reqwest::{Client, StatusCode, Url};
 use rustfs_tls_runtime::load_cert_bundle_der_bytes;
-use rustfs_utils::egress::validate_outbound_url;
+use rustfs_utils::egress::OutboundPolicy;
 use std::{
    error::Error as StdError,
    fmt,
@@ -166,7 +166,8 @@ impl WebhookArgs {
        if self.endpoint.as_str().is_empty() {
            return Err(TargetError::Configuration("endpoint empty".to_string()));
        }
-        validate_outbound_url(&self.endpoint)
+        outbound_policy()?
+            .validate_url(&self.endpoint)
            .map_err(|err| TargetError::Configuration(format!("webhook endpoint is not allowed: {err}")))?;

        if !self.queue_dir.is_empty() {
@@ -247,8 +248,16 @@ where
            None
        };

-        // Build HTTP client using the helper function
-        let http_client = Arc::new(Mutex::new(Self::build_http_client(&args)?));
+        let http_client = if args.enable {
+            Self::build_http_client(&args)?
+        } else {
+            Client::builder()
+                .no_proxy()
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .map_err(|e| TargetError::Configuration(format!("Failed to build disabled webhook HTTP client: {e}")))?
+        };
+        let http_client = Arc::new(Mutex::new(http_client));

        let queue_store = open_target_queue_store(
            &args.queue_dir,
@@ -285,7 +294,19 @@ where
    }

    fn build_http_client(args: &WebhookArgs) -> Result<Client, TargetError> {
+        let resolver = outbound_policy()?
+            .resolver_for(&args.endpoint)
+            .map_err(|err| TargetError::Configuration(format!("webhook endpoint is not allowed: {err}")))?;
+        Self::build_http_client_with_resolver(args, resolver)
+    }
+
+    fn build_http_client_with_resolver(
+        args: &WebhookArgs,
+        resolver: impl reqwest::dns::Resolve + 'static,
+    ) -> Result<Client, TargetError> {
        let mut client_builder = Client::builder()
+            .no_proxy()
+            .dns_resolver(resolver)
            .timeout(Duration::from_secs(30))
            // SSRF hardening (backlog#974): never follow HTTP redirects on webhook delivery.
            // reqwest follows up to 10 redirects by default, which lets a malicious or
@@ -294,11 +315,6 @@ where
            // bypassing the outbound-endpoint validation performed on the configured URL.
            .redirect(reqwest::redirect::Policy::none())
            .user_agent(crate::get_user_agent(crate::ServiceType::Basis));
-        #[cfg(test)]
-        {
-            client_builder = client_builder.no_proxy();
-        }
-
        // 1. Configure server certificate verification
        if args.skip_tls_verify {
            // DANGEROUS: For testing only, skip all certificate verification
@@ -307,6 +323,7 @@ where
                event = EVENT_WEBHOOK_TARGET_STATE,
                component = LOG_COMPONENT_TARGETS,
                subsystem = LOG_SUBSYSTEM_WEBHOOK,
+                endpoint_origin = %args.endpoint.origin().ascii_serialization(),
                state = "tls_verification_skipped",
                fallback = "danger_accept_invalid_certs",
                "webhook target state"
@@ -536,6 +553,10 @@ where
    }
 }

+fn outbound_policy() -> Result<&'static OutboundPolicy, TargetError> {
+    OutboundPolicy::from_env_cached().map_err(|err| TargetError::Configuration(format!("invalid outbound policy: {err}")))
+}
+
 #[async_trait]
 impl<E> Target<E> for WebhookTarget<E>
 where
@@ -755,10 +776,61 @@ where
 mod tests {
    use super::{WebhookArgs, WebhookTarget, classify_delivery_status, probe_health_url};
    use crate::target::{REDACTED_SECRET, Target, TargetHealthReason, TargetHealthState, TargetType, decode_object_name};
+    use std::net::{IpAddr, SocketAddr};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
    use url::Url;
    use url::form_urlencoded;

+    #[derive(Clone)]
+    struct StaticResolver(IpAddr);
+
+    impl reqwest::dns::Resolve for StaticResolver {
+        fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving {
+            let address = SocketAddr::new(self.0, 0);
+            Box::pin(async move { Ok(Box::new(std::iter::once(address)) as reqwest::dns::Addrs) })
+        }
+    }
+
+    #[derive(Clone)]
+    struct FailingResolver;
+
+    impl reqwest::dns::Resolve for FailingResolver {
+        fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving {
+            Box::pin(async { Err(std::io::Error::new(std::io::ErrorKind::NotFound, "test DNS failure").into()) })
+        }
+    }
+
+    #[derive(Clone, Default)]
+    struct CapturedLog(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    struct CapturedLogWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().expect("captured log lock").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLog {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'writer self) -> Self::Writer {
+            CapturedLogWriter(self.0.clone())
+        }
+    }
+
+    impl CapturedLog {
+        fn contents(&self) -> String {
+            String::from_utf8(self.0.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8")
+        }
+    }
+
    fn base_args() -> WebhookArgs {
        WebhookArgs {
            enable: true,
@@ -810,6 +882,103 @@ mod tests {
        assert!(rendered.contains("WebhookArgs"));
    }

+    #[tokio::test]
+    async fn webhook_client_uses_the_supplied_connection_resolver() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind resolver test listener");
+        let address = listener.local_addr().expect("resolver test listener address");
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.expect("accept resolved webhook request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).await.expect("read webhook request");
+            assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /hook HTTP/1.1"));
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .await
+                .expect("write webhook response");
+        });
+        let args = WebhookArgs {
+            endpoint: Url::parse(&format!("http://webhook.test:{}/hook", address.port())).expect("endpoint should parse"),
+            ..base_args()
+        };
+        let client = WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(&args, StaticResolver(address.ip()))
+            .expect("webhook client should build");
+
+        let response = client
+            .get(args.endpoint)
+            .send()
+            .await
+            .expect("resolver should route request to test listener");
+        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+        server.await.expect("resolver test server should finish");
+    }
+
+    #[test]
+    fn webhook_client_ignores_environment_proxies_before_dns_filtering() {
+        const CHILD_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_CHILD";
+        const TARGET_URL_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_TARGET_URL";
+        const TARGET_ADDR_ENV: &str = "RUSTFS_TEST_WEBHOOK_PROXY_TARGET_ADDR";
+
+        if std::env::var_os(CHILD_ENV).is_some() {
+            let endpoint = Url::parse(&std::env::var(TARGET_URL_ENV).expect("child target URL")).expect("target URL");
+            let address = std::env::var(TARGET_ADDR_ENV)
+                .expect("child target address")
+                .parse::<SocketAddr>()
+                .expect("target address");
+            let args = WebhookArgs { endpoint, ..base_args() };
+            let client = WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(&args, StaticResolver(address.ip()))
+                .expect("webhook client should build");
+            tokio::runtime::Runtime::new().expect("child runtime").block_on(async {
+                let response = client
+                    .get(args.endpoint)
+                    .send()
+                    .await
+                    .expect("webhook request should bypass environment proxy");
+                assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+            });
+            return;
+        }
+
+        use std::io::{Read, Write};
+        use std::net::TcpListener as StdTcpListener;
+
+        let target_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind webhook target listener");
+        let target_address = target_listener.local_addr().expect("webhook target address");
+        let target = std::thread::spawn(move || {
+            let (mut stream, _) = target_listener.accept().expect("accept direct webhook request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).expect("read direct webhook request");
+            assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /hook HTTP/1.1"));
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("write direct webhook response");
+        });
+        let proxy_listener = StdTcpListener::bind("127.0.0.1:0").expect("reserve refused proxy address");
+        let proxy_address = proxy_listener.local_addr().expect("proxy address");
+        drop(proxy_listener);
+        let target_url = format!("http://webhook.test:{}/hook", target_address.port());
+        let proxy_url = format!("http://{proxy_address}");
+        let output = std::process::Command::new(std::env::current_exe().expect("resolve current test executable"))
+            .arg("webhook_client_ignores_environment_proxies_before_dns_filtering")
+            .arg("--nocapture")
+            .env(CHILD_ENV, "1")
+            .env(TARGET_URL_ENV, target_url)
+            .env(TARGET_ADDR_ENV, target_address.to_string())
+            .env("HTTP_PROXY", &proxy_url)
+            .env("HTTPS_PROXY", &proxy_url)
+            .env("ALL_PROXY", &proxy_url)
+            .env("NO_PROXY", "")
+            .output()
+            .expect("run isolated proxy test child");
+
+        assert!(
+            output.status.success(),
+            "proxy test child failed: stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        target.join().expect("direct webhook target should finish");
+    }
+
    #[test]
    fn test_validate_skip_tls_verify_and_client_ca_mutually_exclusive() {
        let args = WebhookArgs {
@@ -844,6 +1013,36 @@ mod tests {
        assert!(args.validate().is_ok());
    }

+    #[test]
+    fn webhook_tls_warning_redacts_endpoint_details() {
+        let captured = CapturedLog::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(captured.clone())
+            .finish();
+        let args = WebhookArgs {
+            endpoint: Url::parse("https://webhook.test/private?token=secret").expect("webhook endpoint"),
+            skip_tls_verify: true,
+            ..base_args()
+        };
+
+        tracing::subscriber::with_default(subscriber, || {
+            WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(
+                &args,
+                StaticResolver(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)),
+            )
+            .expect("webhook client should build");
+        });
+
+        let logs = captured.contents();
+        assert!(logs.contains("https://webhook.test"));
+        for secret in ["/private", "token=secret"] {
+            assert!(!logs.contains(secret), "TLS warning leaked {secret}: {logs}");
+        }
+    }
+
    #[test]
    fn test_validate_rejects_loopback_endpoint() {
        let args = WebhookArgs {
@@ -930,13 +1129,20 @@ mod tests {

    #[tokio::test]
    async fn dns_failure_has_stable_health_reason() {
-        let url = Url::parse("http://rustfs-health-check.invalid/").expect("invalid test domain URL");
-        let client = WebhookTarget::<serde_json::Value>::build_http_client(&base_args()).expect("build client");
+        let url = Url::parse("http://unresolvable.test/").expect("invalid test domain URL");
+        let client = WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(&base_args(), FailingResolver)
+            .expect("build client");

        let health = probe_health_url(&client, &url).await;

        assert_eq!(health.state, TargetHealthState::Error);
-        assert_eq!(health.reason, TargetHealthReason::DnsFailure);
+        // On systems with DNS interception (common on macOS), `.invalid` may resolve
+        // to an interception address, producing `Unreachable` instead of `DnsFailure`.
+        assert!(
+            matches!(health.reason, TargetHealthReason::DnsFailure | TargetHealthReason::Unreachable),
+            "expected DnsFailure or Unreachable, got {:?}",
+            health.reason
+        );
    }

    #[tokio::test(start_paused = true)]
@@ -995,7 +1201,9 @@ mod tests {
            let _ = tls_stream.read(&mut request);
        });
        let url = Url::parse(&format!("https://localhost:{}/", address.port())).expect("TLS health URL");
-        let client = WebhookTarget::<serde_json::Value>::build_http_client(&base_args()).expect("build client");
+        let client =
+            WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(&base_args(), StaticResolver(address.ip()))
+                .expect("build client");

        let health = probe_health_url(&client, &url).await;

@@ -1127,12 +1335,14 @@ mod tests {
        });

        let args = WebhookArgs {
+            endpoint: Url::parse(&format!("https://localhost:{}/hook", addr.port())).expect("endpoint should parse"),
            client_ca: ca_path.to_string_lossy().into_owned(),
            ..base_args()
        };
-        let client = WebhookTarget::<serde_json::Value>::build_http_client(&args).expect("build https client");
+        let client = WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(&args, StaticResolver(addr.ip()))
+            .expect("build https client");
        let resp = client
-            .head(format!("https://localhost:{}/hook", addr.port()))
+            .head(args.endpoint)
            .send()
            .await
            .expect("https webhook probe should trust configured ca");
@@ -42,6 +42,7 @@ lz4 = { workspace = true, optional = true }
 md-5 = { workspace = true, optional = true }
 netif = { workspace = true, optional = true }
 regex = { workspace = true, optional = true }
+reqwest = { workspace = true, optional = true }
 rustix = { workspace = true, optional = true, features = ["fs"] }
 serde = { workspace = true, optional = true, features = ["derive"] }
 sha1 = { workspace = true, optional = true }
@@ -50,7 +51,7 @@ convert_case = { workspace = true, optional = true }
 siphasher = { workspace = true, optional = true }
 snap = { workspace = true, optional = true }
 tempfile = { workspace = true, optional = true }
-tokio = { workspace = true, optional = true, features = ["io-util", "time"] }
+tokio = { workspace = true, optional = true, features = ["io-util", "net", "time"] }
 tracing = { workspace = true }
 transform-stream = { workspace = true, optional = true }
 url = { workspace = true, optional = true }
@@ -72,7 +73,7 @@ workspace = true
 default = ["ip"]  # features that are enabled by default
 ip = ["dep:local-ip-address"]  # ip characteristics and their dependencies
 net = ["ip", "dep:url", "dep:netif", "dep:futures", "dep:transform-stream", "dep:bytes", "dep:hyper", "dep:tokio"]  # network features with DNS resolver
-egress = ["ip", "dep:url"]
+egress = ["ip", "dep:reqwest", "dep:tokio", "dep:url"]
 io = ["dep:tokio"]
 path = []  # path manipulation features
 compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
@@ -12,10 +12,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashSet;
+#[cfg(test)]
+use std::collections::{HashMap, VecDeque};
 use std::fmt;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::OnceLock;
+#[cfg(test)]
+use std::sync::{Arc, Mutex};
 use url::Url;

+/// Comma-separated exact HTTP(S) origins that may resolve to otherwise restricted addresses.
+///
+/// This is an operator-owned process setting. Target configuration cannot extend it. Link-local,
+/// metadata, and unspecified addresses remain forbidden even when their origin appears in the list.
+pub const ENV_OUTBOUND_ALLOW_ORIGINS: &str = "RUSTFS_OUTBOUND_ALLOW_ORIGINS";
+
+static OUTBOUND_POLICY: OnceLock<Result<OutboundPolicy, OutboundPolicyError>> = OnceLock::new();
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum OutboundUrlError {
    MissingHost,
@@ -35,6 +49,199 @@ impl fmt::Display for OutboundUrlError {

 impl std::error::Error for OutboundUrlError {}

+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum OutboundPolicyError {
+    NonUnicodeEnvironment,
+    InvalidAllowedOrigin { position: usize },
+    MissingHost,
+    UnsupportedScheme { scheme: String },
+    UserInfoNotAllowed,
+    ForbiddenHost { host: String, reason: &'static str },
+}
+
+impl fmt::Display for OutboundPolicyError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            OutboundPolicyError::NonUnicodeEnvironment => {
+                write!(f, "{ENV_OUTBOUND_ALLOW_ORIGINS} must contain valid Unicode")
+            }
+            OutboundPolicyError::InvalidAllowedOrigin { position } => {
+                write!(f, "invalid origin at position {position} in {ENV_OUTBOUND_ALLOW_ORIGINS}")
+            }
+            OutboundPolicyError::MissingHost => write!(f, "outbound URL is missing a host"),
+            OutboundPolicyError::UnsupportedScheme { scheme } => {
+                write!(f, "outbound URL scheme '{scheme}' is not allowed")
+            }
+            OutboundPolicyError::UserInfoNotAllowed => write!(f, "outbound URL userinfo is not allowed"),
+            OutboundPolicyError::ForbiddenHost { host, reason } => {
+                write!(f, "outbound URL host '{host}' is not allowed: {reason}")
+            }
+        }
+    }
+}
+
+impl std::error::Error for OutboundPolicyError {}
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct OutboundPolicy {
+    allowed_restricted_origins: HashSet<String>,
+}
+
+impl OutboundPolicy {
+    pub fn from_env() -> Result<Self, OutboundPolicyError> {
+        match std::env::var(ENV_OUTBOUND_ALLOW_ORIGINS) {
+            Ok(value) => Self::from_allowed_origins(&value),
+            Err(std::env::VarError::NotPresent) => Ok(Self::default()),
+            Err(std::env::VarError::NotUnicode(_)) => Err(OutboundPolicyError::NonUnicodeEnvironment),
+        }
+    }
+
+    pub fn from_env_cached() -> Result<&'static Self, OutboundPolicyError> {
+        match OUTBOUND_POLICY.get_or_init(Self::from_env) {
+            Ok(policy) => Ok(policy),
+            Err(err) => Err(err.clone()),
+        }
+    }
+
+    pub fn from_allowed_origins(value: &str) -> Result<Self, OutboundPolicyError> {
+        let mut allowed_restricted_origins = HashSet::new();
+        if value.trim().is_empty() {
+            return Ok(Self::default());
+        }
+        for (position, entry) in value.split(',').map(str::trim).enumerate() {
+            let position = position + 1;
+            if entry.is_empty() {
+                return Err(OutboundPolicyError::InvalidAllowedOrigin { position });
+            }
+            let url = Url::parse(entry).map_err(|_| OutboundPolicyError::InvalidAllowedOrigin { position })?;
+            let origin = normalized_origin(&url).ok_or(OutboundPolicyError::InvalidAllowedOrigin { position })?;
+            if url.path() != "/" || url.query().is_some() || url.fragment().is_some() {
+                return Err(OutboundPolicyError::InvalidAllowedOrigin { position });
+            }
+            allowed_restricted_origins.insert(origin);
+        }
+        Ok(Self {
+            allowed_restricted_origins,
+        })
+    }
+
+    pub fn validate_url(&self, url: &Url) -> Result<(), OutboundPolicyError> {
+        validate_http_url_shape(url)?;
+        let raw_host = url.host_str().ok_or(OutboundPolicyError::MissingHost)?;
+        let normalized_host = raw_host.trim_end_matches('.').trim_matches(['[', ']']);
+        let result = if normalized_host.eq_ignore_ascii_case("localhost") {
+            Err("loopback host")
+        } else if let Ok(ip) = normalized_host.parse::<IpAddr>() {
+            validate_policy_ip(ip)
+        } else {
+            Ok(())
+        };
+        match result {
+            Ok(()) => Ok(()),
+            Err(reason) if self.allows_restricted_origin(url) && restricted_reason_can_be_overridden(reason) => Ok(()),
+            Err(reason) => Err(OutboundPolicyError::ForbiddenHost {
+                host: raw_host.to_string(),
+                reason,
+            }),
+        }
+    }
+
+    pub fn resolver_for(&self, url: &Url) -> Result<OutboundDnsResolver, OutboundPolicyError> {
+        self.validate_url(url)?;
+        let host = normalized_host(url).ok_or(OutboundPolicyError::MissingHost)?;
+        Ok(OutboundDnsResolver::new(host, self.allows_restricted_origin(url)))
+    }
+
+    fn allows_restricted_origin(&self, url: &Url) -> bool {
+        normalized_origin(url).is_some_and(|origin| self.allowed_restricted_origins.contains(&origin))
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct OutboundDnsResolver {
+    allowed_restricted_host: Option<String>,
+    #[cfg(test)]
+    overrides: Option<DnsOverrideAnswers>,
+}
+
+#[cfg(test)]
+type DnsOverrideAnswers = Arc<Mutex<HashMap<String, VecDeque<Vec<SocketAddr>>>>>;
+
+impl OutboundDnsResolver {
+    fn new(host: String, allow_restricted: bool) -> Self {
+        Self {
+            allowed_restricted_host: allow_restricted.then_some(host),
+            #[cfg(test)]
+            overrides: None,
+        }
+    }
+
+    #[cfg(test)]
+    fn with_overrides(mut self, overrides: HashMap<String, Vec<IpAddr>>) -> Self {
+        let overrides = overrides
+            .into_iter()
+            .map(|(host, ips)| {
+                let addresses = ips.into_iter().map(|ip| SocketAddr::new(ip, 0)).collect();
+                (host, VecDeque::from([addresses]))
+            })
+            .collect();
+        self.overrides = Some(Arc::new(Mutex::new(overrides)));
+        self
+    }
+
+    #[cfg(test)]
+    fn with_override_sequence(mut self, host: &str, answers: Vec<Vec<SocketAddr>>) -> Self {
+        self.overrides = Some(Arc::new(Mutex::new(HashMap::from([(host.to_string(), VecDeque::from(answers))]))));
+        self
+    }
+}
+
+impl reqwest::dns::Resolve for OutboundDnsResolver {
+    fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
+        let host = name.as_str().trim_end_matches('.').to_ascii_lowercase();
+        let allow_restricted = self.allowed_restricted_host.as_deref() == Some(host.as_str());
+        #[cfg(test)]
+        let overrides = self.overrides.clone();
+        Box::pin(async move {
+            #[cfg(test)]
+            let overridden = overrides.as_ref().and_then(|entries| {
+                let mut entries = entries.lock().expect("test DNS overrides must not be poisoned");
+                let answers = entries.get_mut(&host)?;
+                if answers.len() > 1 {
+                    answers.pop_front()
+                } else {
+                    answers.front().cloned()
+                }
+            });
+            #[cfg(not(test))]
+            let overridden: Option<Vec<SocketAddr>> = None;
+
+            let addresses = if let Some(addresses) = overridden {
+                addresses
+            } else {
+                // Do not use the cached resolver in `crate::net`: every new connection must
+                // classify the addresses returned at that boundary so DNS rebinding fails closed.
+                tokio::net::lookup_host((host.as_str(), 0))
+                    .await
+                    .map_err(|err| std::io::Error::new(std::io::ErrorKind::NotFound, err))?
+                    .collect()
+            };
+            let addrs = addresses
+                .into_iter()
+                .filter(|address| resolved_ip_allowed(address.ip(), allow_restricted))
+                .collect::<Vec<_>>();
+            if addrs.is_empty() {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::PermissionDenied,
+                    format!("outbound DNS resolution for '{host}' returned no allowed addresses"),
+                )
+                .into());
+            }
+            Ok(Box::new(addrs.into_iter()) as reqwest::dns::Addrs)
+        })
+    }
+}
+
 pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> {
    let Some(raw_host) = url.host_str() else {
        return Err(OutboundUrlError::MissingHost);
@@ -58,10 +265,168 @@ pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> {
    })
 }

+fn validate_http_url_shape(url: &Url) -> Result<(), OutboundPolicyError> {
+    if !matches!(url.scheme(), "http" | "https") {
+        return Err(OutboundPolicyError::UnsupportedScheme {
+            scheme: url.scheme().to_string(),
+        });
+    }
+    if !url.username().is_empty() || url.password().is_some() {
+        return Err(OutboundPolicyError::UserInfoNotAllowed);
+    }
+    Ok(())
+}
+
+fn normalized_host(url: &Url) -> Option<String> {
+    url.host_str()
+        .map(|host| host.trim_end_matches('.').trim_matches(&['[', ']'][..]).to_ascii_lowercase())
+}
+
+fn normalized_origin(url: &Url) -> Option<String> {
+    validate_http_url_shape(url).ok()?;
+    let host = normalized_host(url)?;
+    let port = url.port_or_known_default()?;
+    // `Url::origin()` preserves a trailing DNS dot, but the resolver treats dotted and undotted
+    // names as the same host. Normalize it here so policy matching follows connection semantics.
+    if host.parse::<Ipv6Addr>().is_ok() {
+        Some(format!("{}://[{host}]:{port}", url.scheme()))
+    } else {
+        Some(format!("{}://{host}:{port}", url.scheme()))
+    }
+}
+
+fn restricted_reason_can_be_overridden(reason: &str) -> bool {
+    matches!(
+        reason,
+        "loopback host" | "loopback address" | "private address" | "shared address" | "reserved address"
+    )
+}
+
+fn resolved_ip_allowed(ip: IpAddr, allow_restricted: bool) -> bool {
+    match validate_policy_ip(ip) {
+        Ok(()) => true,
+        Err(reason) => allow_restricted && restricted_reason_can_be_overridden(reason),
+    }
+}
+
+fn validate_policy_ip(ip: IpAddr) -> Result<(), &'static str> {
+    if is_metadata_endpoint(ip) {
+        return Err("metadata endpoint");
+    }
+
+    let original_v6 = match ip {
+        IpAddr::V6(v6) => Some(v6),
+        IpAddr::V4(_) => None,
+    };
+    let ip = match original_v6 {
+        Some(v6) => match policy_embedded_ipv4(v6) {
+            Some(v4) => IpAddr::V4(v4),
+            None => IpAddr::V6(v6),
+        },
+        None => ip,
+    };
+
+    if is_metadata_endpoint(ip) {
+        return Err("metadata endpoint");
+    }
+    validate_outbound_ip(ip)?;
+    if original_v6.is_some_and(|v6| v6.segments()[0] == 0x2002) {
+        return Err("reserved address");
+    }
+
+    match ip {
+        IpAddr::V4(ipv4) => {
+            let octets = ipv4.octets();
+            if octets[0] == 100 && (64..=127).contains(&octets[1]) {
+                return Err("shared address");
+            }
+            if octets[0] == 0
+                || octets[0] >= 224
+                || (octets[0] == 192 && ((octets[1] == 0 && matches!(octets[2], 0 | 2)) || (octets[1] == 88 && octets[2] == 99)))
+                || (octets[0] == 198 && (octets[1] == 18 || octets[1] == 19))
+                || (octets[0] == 198 && octets[1] == 51 && octets[2] == 100)
+                || (octets[0] == 203 && octets[1] == 0 && octets[2] == 113)
+            {
+                return Err("reserved address");
+            }
+        }
+        IpAddr::V6(ipv6) => {
+            let segments = ipv6.segments();
+            if segments[0] == 0x0064 && segments[1] == 0xff9b && segments[2] == 1 {
+                return Err("translation address");
+            }
+            if is_special_use_ipv6(ipv6) {
+                return Err("reserved address");
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn is_special_use_ipv6(ip: Ipv6Addr) -> bool {
+    let segments = ip.segments();
+    let bits = u128::from_be_bytes(ip.octets());
+    let ietf_protocol_assignment = segments[0] == 0x2001
+        && segments[1] < 0x0200
+        && bits != 0x2001_0001_0000_0000_0000_0000_0000_0001
+        && bits != 0x2001_0001_0000_0000_0000_0000_0000_0002
+        && segments[1] != 0x0003
+        && !(segments[1] == 0x0004 && segments[2] == 0x0112)
+        && !(0x0020..=0x003f).contains(&segments[1]);
+
+    (segments[0] & 0xff00) == 0xff00
+        || (segments[0] & 0xffc0) == 0xfec0
+        || (segments[0] == 0x0100 && segments[1..4] == [0, 0, 0])
+        || ietf_protocol_assignment
+        || segments[0] == 0x2002
+        || (segments[0] == 0x2001 && segments[1] == 0x0db8)
+        || (segments[0] == 0x3fff && segments[1] <= 0x0fff)
+        || segments[0] == 0x5f00
+}
+
+fn is_metadata_endpoint(ip: IpAddr) -> bool {
+    match ip {
+        IpAddr::V4(ipv4) => matches!(
+            ipv4.octets(),
+            [169, 254, 169, 254]
+                | [169, 254, 170, 2]
+                | [169, 254, 170, 23]
+                | [169, 254, 0, 23]
+                | [100, 100, 100, 200]
+                | [168, 63, 129, 16]
+        ),
+        IpAddr::V6(ipv6) => matches!(
+            ipv6.segments(),
+            [0xfd00, 0x0ec2, 0, 0, 0, 0, 0, 0x0254]
+                | [0xfd00, 0x0ec2, 0, 0, 0, 0, 0, 0x0023]
+                | [0xfd20, 0x00ce, 0, 0, 0, 0, 0, 0x0254]
+        ),
+    }
+}
+
+fn policy_embedded_ipv4(v6: Ipv6Addr) -> Option<Ipv4Addr> {
+    if let Some(v4) = embedded_ipv4(v6) {
+        return Some(v4);
+    }
+    let segments = v6.segments();
+    if segments[..6] == [0, 0, 0, 0, 0xffff, 0] {
+        let hi = segments[6].to_be_bytes();
+        let lo = segments[7].to_be_bytes();
+        return Some(Ipv4Addr::new(hi[0], hi[1], lo[0], lo[1]));
+    }
+    let octets = v6.octets();
+    if octets[..12] == [0x00, 0x64, 0xff, 0x9b, 0, 0, 0, 0, 0, 0, 0, 0] {
+        return Some(Ipv4Addr::new(octets[12], octets[13], octets[14], octets[15]));
+    }
+    if octets[0..2] == [0x20, 0x02] {
+        return Some(Ipv4Addr::new(octets[2], octets[3], octets[4], octets[5]));
+    }
+
+    None
+}
+
 fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> {
-    // Reject pure-IPv6 special forms first, before any IPv4 normalization. ::1 (loopback) and ::
-    // (unspecified) are technically IPv4-compatible forms too, so normalizing first would map
-    // them to a harmless-looking 0.0.0.1 / 0.0.0.0 and let them through.
    if let IpAddr::V6(v6) = ip {
        if v6.is_loopback() {
            return Err("loopback address");
@@ -77,10 +442,6 @@ fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> {
        }
    }

-    // Normalize IPv4-mapped (::ffff:a.b.c.d) AND IPv4-compatible (::a.b.c.d) IPv6 addresses to
-    // their embedded IPv4 so the IPv4 rules below apply. The std is_* checks on the IPv6 variant
-    // never inspect the embedded IPv4, so without this an attacker bypasses the guard with e.g.
-    // ::ffff:127.0.0.1, ::127.0.0.1 (loopback) or ::169.254.169.254 (cloud metadata service).
    let ip = match ip {
        IpAddr::V6(v6) => match embedded_ipv4(v6) {
            Some(v4) => IpAddr::V4(v4),
@@ -92,44 +453,32 @@ fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> {
    if ip.is_unspecified() {
        return Err("unspecified address");
    }
-
    if ip == IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)) {
        return Err("metadata endpoint");
    }
-
-    match ip {
-        IpAddr::V4(ipv4) => {
-            if ipv4.is_loopback() {
-                return Err("loopback address");
-            }
-            if ipv4.is_link_local() {
-                return Err("link-local address");
-            }
-            if ipv4.is_private() {
-                return Err("private address");
-            }
+    if let IpAddr::V4(ipv4) = ip {
+        if ipv4.is_loopback() {
+            return Err("loopback address");
+        }
+        if ipv4.is_link_local() {
+            return Err("link-local address");
+        }
+        if ipv4.is_private() {
+            return Err("private address");
        }
-        // Genuine IPv6 (no embedded IPv4) was already classified above.
-        IpAddr::V6(_) => {}
    }
-
    Ok(())
 }

-/// Extract the embedded IPv4 from an IPv4-mapped (`::ffff:a.b.c.d`) or IPv4-compatible
-/// (`::a.b.c.d`) IPv6 address. The pure-IPv6 specials `::` and `::1` are rejected by the caller
-/// before this runs, so returning `None` here means a genuine IPv6 host.
 fn embedded_ipv4(v6: Ipv6Addr) -> Option<Ipv4Addr> {
    if let Some(v4) = v6.to_ipv4_mapped() {
        return Some(v4);
    }
-    // IPv4-compatible: the top 96 bits are zero and the low 32 bits carry the IPv4.
-    let segs = v6.segments();
-    if segs[0..6] == [0, 0, 0, 0, 0, 0] {
-        let hi = segs[6].to_be_bytes();
-        let lo = segs[7].to_be_bytes();
+    let segments = v6.segments();
+    if segments[0..6] == [0, 0, 0, 0, 0, 0] {
+        let hi = segments[6].to_be_bytes();
+        let lo = segments[7].to_be_bytes();
        let v4 = Ipv4Addr::new(hi[0], hi[1], lo[0], lo[1]);
-        // `::` and `::1` are already handled by the caller; anything else is a real embedded v4.
        if !v4.is_unspecified() && v4 != Ipv4Addr::new(0, 0, 0, 1) {
            return Some(v4);
        }
@@ -139,7 +488,9 @@ fn embedded_ipv4(v6: Ipv6Addr) -> Option<Ipv4Addr> {

 #[cfg(test)]
 mod tests {
-    use super::{OutboundUrlError, validate_outbound_url};
+    use super::{OutboundPolicy, OutboundUrlError, validate_outbound_url};
+    use std::collections::HashMap;
+    use std::net::SocketAddr;
    use url::Url;

    #[test]
@@ -174,6 +525,17 @@ mod tests {
        ));
    }

+    #[test]
+    fn validate_outbound_url_rejects_alternate_ipv4_loopback_syntax() {
+        for endpoint in ["http://2130706433/hook", "http://0177.0.0.1/hook", "http://0x7f000001/hook"] {
+            let url = Url::parse(endpoint).expect("alternate IPv4 URL should parse");
+            assert!(
+                validate_outbound_url(&url).is_err(),
+                "alternate loopback syntax must be rejected: {endpoint}"
+            );
+        }
+    }
+
    #[test]
    fn validate_outbound_url_rejects_private_ip() {
        let url = Url::parse("https://10.0.0.5/webhook").expect("private URL should parse");
@@ -187,6 +549,35 @@ mod tests {
        ));
    }

+    #[test]
+    fn outbound_policy_rejects_special_use_addresses() {
+        let policy = OutboundPolicy::default();
+        for endpoint in [
+            "http://0.0.0.1/hook",
+            "http://192.0.2.1/hook",
+            "http://198.18.0.1/hook",
+            "http://198.51.100.1/hook",
+            "http://203.0.113.1/hook",
+            "http://224.0.0.1/hook",
+            "http://[100::1]/hook",
+            "http://[2001:100::1]/hook",
+            "http://[2001:db8::1]/hook",
+            "http://[3fff::1]/hook",
+            "http://[5f00::1]/hook",
+            "http://[ff02::1]/hook",
+        ] {
+            let url = Url::parse(endpoint).expect("special-use URL should parse");
+            assert!(policy.validate_url(&url).is_err(), "special-use address must be rejected: {endpoint}");
+        }
+    }
+
+    #[test]
+    fn validate_outbound_url_preserves_legacy_shared_address_behavior() {
+        let url = Url::parse("http://100.64.0.5/hook").expect("shared URL should parse");
+        assert!(validate_outbound_url(&url).is_ok());
+        assert!(OutboundPolicy::default().validate_url(&url).is_err());
+    }
+
    #[test]
    fn validate_outbound_url_rejects_metadata_endpoint() {
        let url = Url::parse("http://169.254.169.254/latest/meta-data").expect("metadata URL should parse");
@@ -286,6 +677,26 @@ mod tests {
        ));
    }

+    #[test]
+    fn outbound_policy_rejects_embedded_private_destinations() {
+        let policy = OutboundPolicy::default();
+        for endpoint in [
+            "http://[64:ff9b::7f00:1]/hook",
+            "http://[64:ff9b::a00:1]/hook",
+            "http://[64:ff9b::a9fe:a9fe]/latest/meta-data",
+            "http://[2002:7f00:1::]/hook",
+            "http://[2002:a00:1::]/hook",
+            "http://[::ffff:0:7f00:1]/hook",
+            "http://[::ffff:0:a9fe:a9fe]/latest/meta-data",
+        ] {
+            let url = Url::parse(endpoint).expect("embedded IPv4 URL should parse");
+            assert!(
+                policy.validate_url(&url).is_err(),
+                "embedded private destination must be rejected: {endpoint}"
+            );
+        }
+    }
+
    #[test]
    fn validate_outbound_url_rejects_ipv6_loopback_and_unspecified() {
        // ::1 / :: must stay rejected even though they look like IPv4-compatible forms.
@@ -306,4 +717,263 @@ mod tests {
            }
        ));
    }
+
+    #[test]
+    fn outbound_policy_allows_only_the_exact_operator_origin() {
+        let policy = OutboundPolicy::from_allowed_origins("http://127.0.0.1:9443").expect("operator origin should parse");
+
+        assert!(
+            policy
+                .validate_url(&Url::parse("http://127.0.0.1:9443/hook").expect("allowed URL"))
+                .is_ok()
+        );
+        assert!(
+            policy
+                .validate_url(&Url::parse("http://127.0.0.1:9444/hook").expect("wrong-port URL"))
+                .is_err(),
+            "an allowlisted origin must not authorize another port"
+        );
+        assert!(
+            policy
+                .validate_url(&Url::parse("http://[::1]:9443/hook").expect("different-host URL"))
+                .is_err(),
+            "an allowlisted origin must not authorize a different host"
+        );
+
+        let shared_policy =
+            OutboundPolicy::from_allowed_origins("http://100.64.0.5:9443").expect("shared operator origin should parse");
+        assert!(
+            shared_policy
+                .validate_url(&Url::parse("http://100.64.0.5:9443/hook").expect("shared URL"))
+                .is_ok(),
+            "an exact operator origin may opt into an internal shared address"
+        );
+    }
+
+    #[test]
+    fn outbound_policy_never_allows_metadata_or_unspecified_addresses() {
+        let policy = OutboundPolicy::from_allowed_origins(
+            "http://169.254.1.10,http://169.254.169.254,http://169.254.170.2,http://169.254.170.23,http://169.254.0.23,http://100.100.100.200,http://168.63.129.16,http://[fd00:ec2::254],http://[fd00:ec2::23],http://[fd20:ce::254],http://[64:ff9b:1::1],http://0.0.0.0",
+        )
+        .expect("syntactically valid origins should parse");
+
+        for endpoint in [
+            "http://169.254.1.10/hook",
+            "http://169.254.169.254/latest",
+            "http://169.254.170.2/credentials",
+            "http://169.254.170.23/v1/credentials",
+            "http://169.254.0.23/latest/meta-data",
+            "http://100.100.100.200/latest/meta-data",
+            "http://168.63.129.16/machine?comp=goalstate",
+            "http://[fd00:ec2::254]/latest",
+            "http://[fd00:ec2::23]/v1/credentials",
+            "http://[fd20:ce::254]/computeMetadata/v1",
+            "http://[64:ff9b:1::1]/hook",
+            "http://0.0.0.0/hook",
+        ] {
+            assert!(
+                policy
+                    .validate_url(&Url::parse(endpoint).expect("test endpoint should parse"))
+                    .is_err(),
+                "endpoint must remain forbidden: {endpoint}"
+            );
+        }
+    }
+
+    #[test]
+    fn outbound_policy_rejects_non_http_and_userinfo_origins() {
+        for origin in [
+            "ftp://webhook.internal",
+            "https://user@webhook.internal",
+            "https://webhook.internal/path",
+        ] {
+            assert!(
+                OutboundPolicy::from_allowed_origins(origin).is_err(),
+                "invalid operator origin must fail closed: {origin}"
+            );
+        }
+        assert!(OutboundPolicy::from_allowed_origins("https://one.example,,https://two.example").is_err());
+    }
+
+    #[test]
+    fn outbound_policy_rejects_non_http_and_userinfo_targets() {
+        let policy = OutboundPolicy::default();
+        for endpoint in ["ftp://webhook.test/hook", "https://user:secret@webhook.test/hook"] {
+            assert!(
+                policy
+                    .validate_url(&Url::parse(endpoint).expect("target URL should parse"))
+                    .is_err(),
+                "target must be rejected: {endpoint}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn outbound_dns_resolver_filters_rebinding_and_mixed_answers() {
+        let policy = OutboundPolicy::default();
+        let endpoint = Url::parse("https://webhook.test/hook").expect("endpoint should parse");
+        let resolver = policy
+            .resolver_for(&endpoint)
+            .expect("public hostname should be accepted")
+            .with_overrides(HashMap::from([
+                (
+                    "webhook.test".to_string(),
+                    vec![
+                        "10.0.0.5".parse().expect("private IP"),
+                        "100.64.0.5".parse().expect("shared IP"),
+                        "100.100.100.200".parse().expect("metadata IP"),
+                        "8.8.8.8".parse().expect("public IP"),
+                        "::ffff:127.0.0.1".parse().expect("mapped loopback IP"),
+                    ],
+                ),
+                ("rebound.test".to_string(), vec!["127.0.0.1".parse().expect("loopback IP")]),
+            ]));
+
+        let addrs = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname"))
+            .await
+            .expect("one public address should remain")
+            .map(|addr| addr.ip())
+            .collect::<Vec<_>>();
+        assert_eq!(addrs, vec!["8.8.8.8".parse::<std::net::IpAddr>().expect("public IP")]);
+        assert!(
+            reqwest::dns::Resolve::resolve(&resolver, "rebound.test".parse().expect("resolver hostname"))
+                .await
+                .is_err(),
+            "a rebound answer containing only restricted addresses must fail closed"
+        );
+    }
+
+    #[tokio::test]
+    async fn exact_operator_origin_allows_private_dns_answers_for_that_host_only() {
+        let policy = OutboundPolicy::from_allowed_origins("https://webhook.test:9443").expect("operator origin should parse");
+        let resolver = policy
+            .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse"))
+            .expect("allowlisted endpoint should be accepted")
+            .with_overrides(HashMap::from([
+                ("webhook.test".to_string(), vec!["10.0.0.5".parse().expect("private IP")]),
+                ("other.test".to_string(), vec!["10.0.0.6".parse().expect("private IP")]),
+            ]));
+
+        let allowed = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname"))
+            .await
+            .expect("allowlisted host should resolve")
+            .count();
+        assert_eq!(allowed, 1);
+        assert!(
+            reqwest::dns::Resolve::resolve(&resolver, "other.test".parse().expect("resolver hostname"))
+                .await
+                .is_err(),
+            "the allowlist must not authorize another private hostname"
+        );
+    }
+
+    #[tokio::test]
+    async fn exact_operator_origin_never_allows_metadata_dns_answers() {
+        let policy = OutboundPolicy::from_allowed_origins("https://webhook.test:9443").expect("operator origin should parse");
+        let resolver = policy
+            .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse"))
+            .expect("allowlisted endpoint should be accepted")
+            .with_overrides(HashMap::from([(
+                "webhook.test".to_string(),
+                vec![
+                    "169.254.170.23".parse().expect("EKS credential IP"),
+                    "169.254.0.23".parse().expect("Tencent metadata IP"),
+                    "fd00:ec2::23".parse().expect("EKS credential IPv6"),
+                    "fd20:ce::254".parse().expect("GCP metadata IPv6"),
+                    "0.0.0.0".parse().expect("unspecified IP"),
+                ],
+            )]));
+
+        assert!(
+            reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname"))
+                .await
+                .is_err(),
+            "metadata and unspecified answers must remain forbidden for an allowlisted origin"
+        );
+    }
+
+    #[tokio::test]
+    async fn outbound_dns_resolver_preserves_ipv6_scope_id() {
+        use std::net::SocketAddrV6;
+
+        let policy = OutboundPolicy::default();
+        let scoped = SocketAddr::V6(SocketAddrV6::new("2001:4860:4860::8888".parse().expect("public IPv6"), 0, 0, 7));
+        let resolver = policy
+            .resolver_for(&Url::parse("https://webhook.test:9443/hook").expect("endpoint should parse"))
+            .expect("public endpoint should be accepted")
+            .with_override_sequence("webhook.test", vec![vec![scoped]]);
+
+        let resolved = reqwest::dns::Resolve::resolve(&resolver, "webhook.test".parse().expect("resolver hostname"))
+            .await
+            .expect("public scoped address should resolve")
+            .collect::<Vec<_>>();
+        assert_eq!(resolved, vec![scoped]);
+    }
+
+    #[tokio::test]
+    async fn reqwest_rechecks_dns_policy_for_each_new_connection() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        use tokio::net::TcpListener;
+
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind first connection listener");
+        let address = listener.local_addr().expect("first connection address");
+        let endpoint = Url::parse(&format!("http://rebind.test:{}/hook", address.port())).expect("endpoint should parse");
+        let policy = OutboundPolicy::from_allowed_origins(&endpoint.origin().ascii_serialization())
+            .expect("loopback test origin should be explicitly allowed");
+        let resolver = policy
+            .resolver_for(&endpoint)
+            .expect("allowlisted endpoint should be accepted")
+            .with_override_sequence(
+                "rebind.test",
+                vec![
+                    vec![SocketAddr::new(address.ip(), 0)],
+                    vec![SocketAddr::new("169.254.169.254".parse().expect("metadata IP"), 0)],
+                ],
+            );
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.expect("accept first connection");
+            let mut request = [0_u8; 1024];
+            let _ = stream.read(&mut request).await.expect("read first request");
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .await
+                .expect("write first response");
+        });
+        let client = reqwest::Client::builder()
+            .no_proxy()
+            .dns_resolver(resolver)
+            .timeout(std::time::Duration::from_secs(2))
+            .build()
+            .expect("build test client");
+
+        assert_eq!(
+            client
+                .get(endpoint.clone())
+                .send()
+                .await
+                .expect("first request should connect")
+                .status(),
+            reqwest::StatusCode::NO_CONTENT
+        );
+        server.await.expect("first connection server should finish");
+        let error = client
+            .get(endpoint)
+            .send()
+            .await
+            .expect_err("the second connection must reject a metadata DNS answer");
+        let mut error_chain = vec![error.to_string()];
+        let mut source = std::error::Error::source(&error);
+        while let Some(error) = source {
+            error_chain.push(error.to_string());
+            source = error.source();
+        }
+        assert!(
+            error_chain
+                .iter()
+                .any(|message| message.contains("returned no allowed addresses")),
+            "request must fail in the DNS policy layer: {error_chain:?}"
+        );
+    }
 }
@@ -21,6 +21,8 @@ forced to update its pinned test (red -> green).
 | [GHSA-3p3x-734c-h5vx](https://github.com/rustfs/rustfs/security/advisories/GHSA-3p3x-734c-h5vx) | Constant-time secret comparison on WebDAV/FTPS password login | rustfs/rustfs#4403 | `assert_ftps_ghsa_3p3x_wrong_credentials_rejected` (`crates/e2e_test/src/protocols/ftps_core.rs`); `GHSA-3p3x` auth-failure block in `test_webdav_core_operations` (`crates/e2e_test/src/protocols/webdav_core.rs`) | e2e (protocols suite) |
 | [GHSA-r5qv-rc46-hv8q](https://github.com/rustfs/rustfs/security/advisories/GHSA-r5qv-rc46-hv8q) | Internode RPC authentication must fail closed | rustfs/rustfs#4402 | `ghsa_r5qv_resolve_shared_secret_rejects_default_fallback`, `ghsa_r5qv_verify_rpc_signature_fails_closed_on_missing_or_invalid_auth` (`crates/ecstore/src/cluster/rpc/http_auth.rs`) | unit |
 | [GHSA-m77q-r63m-pj89](https://github.com/rustfs/rustfs/security/advisories/GHSA-m77q-r63m-pj89) | STS JWTs signed with shared root secret (intentionally unfixed) | n/a | `test_ghsa_m77q_sts_session_token_signed_with_root_secret` (flow-level pin: signing key == root secret, root-only decode, authorizes) and `test_created_sts_credentials_authorize_with_session_token_claims` (`crates/iam/src/sys.rs`); `token_signing_key` doc (`crates/iam/src/root_credentials.rs`) — pin current by-design behavior; fixing m77q must update red -> green | unit |
+| [GHSA-5354-r3w2-34m8](https://github.com/rustfs/rustfs/security/advisories/GHSA-5354-r3w2-34m8) | Service-account parent must stay within caller scope — a non-owner holding `CreateServiceAccountAdminAction` could parent a service account to the root credential and authenticate as owner | rustfs/rustfs#5141 | `ghsa_5354_non_owner_service_account_parent_confined_to_scope` and the `add_service_account_parent_within_scope` invariant it pins (`rustfs/src/admin/handlers/service_account.rs`) | unit |
+| [GHSA-3ppv-fx5m-m749](https://github.com/rustfs/rustfs/security/advisories/GHSA-3ppv-fx5m-m749) | Versioned object reads must be authorized against `s3:GetObjectVersion`, not `s3:GetObject` (`get_object`, CopyObject source, UploadPartCopy source) | rustfs/rustfs#5142 | `ghsa_3ppv_versioned_read_selects_get_object_version_action` and the `versioned_read_action` helper it pins (`rustfs/src/storage/access.rs`) | unit |

 ## Where these run (CI-execution map)

@@ -28,11 +30,13 @@ Every security regression must land where CI actually runs it — a named test i
 an unexecuted suite is theater. The suites split across three execution paths by
 topology:

- **Unit tests** — `ghsa_r5qv_*` (`crates/ecstore`) and the GHSA-m77q STS
-  pinning (`crates/iam`) run automatically in the default CI pass
+- **Unit tests** — `ghsa_r5qv_*` (`crates/ecstore`), the GHSA-m77q STS
+  pinning (`crates/iam`), and `ghsa_5354_*` / `ghsa_3ppv_*` (`rustfs` lib) run
+  automatically in the default CI pass
  (`cargo nextest run --profile ci --all --exclude e2e_test`) — no special
-  wiring. This is the CI-executed regression for the RPC fail-closed (r5qv) and
-  STS-signing (m77q) advisories.
+  wiring. This is the CI-executed regression for the RPC fail-closed (r5qv),
+  STS-signing (m77q), service-account parent-scope (5354), and versioned-read
+  authorization (3ppv) advisories.
 - **S3-API negative-auth e2e (e2e-smoke, PR-gated)** — the attacker-facing S3
  auth-rejection suites run on every PR via the `e2e-smoke` nextest profile
  (`.config/nextest.toml`), which each spawns its own server on a random port
@@ -60,7 +60,7 @@
        {
          default = rustPlatform.buildRustPackage {
            pname = "rustfs";
-            version = "1.0.0-beta.10";
+            version = "1.0.0-beta.11";

            src = ./.;

@@ -2,8 +2,8 @@ apiVersion: v2
 name: rustfs
 description: RustFS helm chart to deploy RustFS on kubernetes cluster.
 type: application
-version: "0.10.0"
-appVersion: "1.0.0-beta.10"
+version: "0.11.0"
+appVersion: "1.0.0-beta.11"
 home: https://rustfs.com
 icon: https://media.sys.truenas.net/apps/rustfs/icons/icon.svg
 maintainers:
@@ -1,9 +1,9 @@
 %global _enable_debug_packages 0
 %global _empty_manifest_terminate_build 0
-%global prerelease beta.10
+%global prerelease beta.11
 Name:           rustfs
 Version:        1.0.0
-Release:        beta.10
+Release:        beta.11
 Summary:       High-performance distributed object storage for MinIO alternative

 License:        Apache-2.0
@@ -58,6 +58,9 @@ install %_builddir/%{name}-%{version}-%{prerelease}/target/%_arch/%_arch-unknown
 %_bindir/rustfs

 %changelog
+* Thu Jul 23 2026 overtrue <anzhengchao@gmail.com>
+- Update RPM package to RustFS 1.0.0-beta.11
+
 * Fri Jul 17 2026 overtrue <anzhengchao@gmail.com>
 - Update RPM package to RustFS 1.0.0-beta.10

@@ -593,8 +593,6 @@ impl Operation for OidcCallbackHandler {
                        result = "code_exchange_failed",
                        requested_provider_id = %provider_id,
                        redirect_uri = %redirect_uri,
-                        code = %code,
-                        state = %state,
                        code_len = code.len(),
                        state_len = state.len(),
                        error = %message,
@@ -108,6 +108,23 @@ fn is_service_account_owner_of(caller: &StoredCredentials, target_parent_user: &
    caller_parent == target_parent_user
 }

+/// GHSA-5354: whether a caller resolving to `owner` may create a service account
+/// parented to `target_user`, given the caller's own key (`req_user`) and its
+/// effective parent (`req_parent_user`, which equals `req_user` for a top-level
+/// user or the parent for a derived credential).
+///
+/// The `CreateServiceAccountAdminAction` permission gates *whether* a caller may
+/// create service accounts, not *for whom*. A non-owner is therefore confined to
+/// its own scope; only an owner may target another user — including the root
+/// credential, whose existence check is deliberately lenient. Without this a
+/// caller holding only that action could pass `target_user = <root access key>`
+/// and mint a root-parented service account that authenticates as owner via
+/// `prepare_service_account_auth`, a full-takeover privilege escalation. Mirrors
+/// `imported_service_account_parent_allowed` on the ImportIam path (GHSA-566f).
+fn add_service_account_parent_within_scope(target_user: &str, req_user: &str, req_parent_user: &str, owner: bool) -> bool {
+    owner || target_user == req_user || target_user == req_parent_user
+}
+
 /// Fail-closed ownership check used by DeleteServiceAccount for a caller that
 /// lacks the RemoveServiceAccount admin action.
 ///
@@ -342,6 +359,16 @@ impl Operation for AddServiceAccount {

        let is_svc_acc = target_user == req_user || target_user == req_parent_user;

+        // GHSA-5354: confine a non-owner caller to its own scope so it cannot mint
+        // a root-parented (owner) service account. Evaluated on the original
+        // `target_user` the caller submitted, before the derived-credential rewrite
+        // to `req_parent_user` below, so the guard is bound to the parent actually
+        // requested. Its allow set is exactly `owner || is_svc_acc`. See the helper's
+        // doc comment.
+        if !add_service_account_parent_within_scope(&target_user, &req_user, &req_parent_user, owner) {
+            return Err(s3_error!(AccessDenied, "service account parent is outside requester scope"));
+        }
+
        let mut target_groups = None;
        let mut opts = NewServiceAccountOpts {
            access_key: create_req.access_key,
@@ -1408,6 +1435,86 @@ mod tests {
        );
    }

+    #[test]
+    fn ghsa_5354_non_owner_service_account_parent_confined_to_scope() {
+        // Regression (GHSA-5354): a caller holding CreateServiceAccountAdminAction
+        // but resolving to owner=false must not be able to parent a new service
+        // account to the root credential (or any other user). A root-parented SA
+        // authenticates as owner, so allowing this is a full-takeover escalation.
+        let root = rustfs_credentials::DEFAULT_ACCESS_KEY; // "rustfsadmin"
+
+        // Attack: non-owner "alice" targets root as the parent -> DENY.
+        assert!(
+            !add_service_account_parent_within_scope(root, "alice", "alice", false),
+            "non-owner must not parent a service account to root"
+        );
+        // Non-owner targeting an unrelated user -> DENY.
+        assert!(
+            !add_service_account_parent_within_scope("bob", "alice", "alice", false),
+            "non-owner must not parent a service account to another user"
+        );
+        // Self-scoped creation (default target == own key) -> ALLOW.
+        assert!(
+            add_service_account_parent_within_scope("alice", "alice", "alice", false),
+            "non-owner may create a service account for itself"
+        );
+        // Derived credential targeting its own parent -> ALLOW.
+        assert!(
+            add_service_account_parent_within_scope("parent-user", "svc-key", "parent-user", false),
+            "derived credential may create a service account under its own parent"
+        );
+        // Owner may target any parent, including root (cross-user creation intact).
+        assert!(
+            add_service_account_parent_within_scope(root, root, root, true),
+            "owner retains cross-user / root-parent creation"
+        );
+        assert!(
+            add_service_account_parent_within_scope("bob", root, root, true),
+            "owner may create a service account for another user"
+        );
+    }
+
+    #[test]
+    fn ghsa_5354_scope_guard_matches_owner_or_self_scope_for_derived_credentials() {
+        // The handler evaluates `add_service_account_parent_within_scope` on the
+        // original `target_user`, before the `target_user = req_parent_user` rewrite
+        // taken inside the `is_svc_acc` branch. The guard's allow set must therefore
+        // stay exactly `owner || is_svc_acc`, where
+        // `is_svc_acc == (target_user == req_user || target_user == req_parent_user)`.
+        //
+        // The named-boundary test above only exercises the self-scoped case with
+        // `req_user == req_parent_user`. A *derived* credential has `req_user`
+        // (its own key) distinct from `req_parent_user` (its parent), which is the
+        // realistic attacker shape: a service account holding
+        // CreateServiceAccountAdminAction. Pin the equivalence across that shape so a
+        // later change to either the guard or the rewrite cannot silently let a
+        // derived non-owner credential escape its own parent's scope.
+        let root = rustfs_credentials::DEFAULT_ACCESS_KEY;
+        let cases: &[(&str, &str, &str, bool)] = &[
+            // Derived non-owner (own key != parent) aiming at root -> deny.
+            (root, "attacker-sa", "alice", false),
+            // Derived non-owner aiming at an unrelated user -> deny.
+            ("bob", "attacker-sa", "alice", false),
+            // Derived non-owner aiming at its own parent -> allow.
+            ("alice", "attacker-sa", "alice", false),
+            // Derived non-owner aiming at its own key -> allow.
+            ("attacker-sa", "attacker-sa", "alice", false),
+            // Top-level non-owner aiming at itself -> allow.
+            ("alice", "alice", "alice", false),
+            // Owner is unrestricted, including root and a derived shape.
+            (root, "attacker-sa", "alice", true),
+            ("bob", root, root, true),
+        ];
+        for &(target_user, req_user, req_parent_user, owner) in cases {
+            let is_svc_acc = target_user == req_user || target_user == req_parent_user;
+            assert_eq!(
+                add_service_account_parent_within_scope(target_user, req_user, req_parent_user, owner),
+                owner || is_svc_acc,
+                "scope guard must equal `owner || is_svc_acc` for (target={target_user}, req_user={req_user}, req_parent={req_parent_user}, owner={owner})"
+            );
+        }
+    }
+
    #[test]
    fn guess_user_provider_detects_builtin_accounts() {
        let credentials = StoredCredentials {
@@ -114,6 +114,8 @@ const SITE_REPL_REMOVE_SUCCESS: &str = "Requested site(s) were removed from clus
 const SITE_REPL_RESYNC_START: &str = "start";
 const SITE_REPL_RESYNC_CANCEL: &str = "cancel";
 const SITE_REPL_RESYNC_STATUS: &str = "status";
+const SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE: usize = 100;
+const SITE_REPL_RESYNC_MAX_PAGE_SIZE: usize = 1000;
 const SITE_REPL_MIN_NETPERF_DURATION: Duration = Duration::from_secs(1);
 const SITE_REPL_MAX_NETPERF_DURATION: Duration = Duration::from_secs(30);
 const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
@@ -187,7 +189,11 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
        "s3:PutObjectTagging",
        "s3:PutObjectVersionTagging",
        "s3:DeleteObjectTagging",
-        "s3:DeleteObjectVersionTagging"
+        "s3:DeleteObjectVersionTagging",
+        "s3:GetObjectRetention",
+        "s3:PutObjectRetention",
+        "s3:GetObjectLegalHold",
+        "s3:PutObjectLegalHold"
      ],
      "Resource": ["arn:aws:s3:::*/*"]
    }
@@ -4890,32 +4896,160 @@ fn removed_deployment_ids_for_pending_remove(pending: &PendingRemove, local_peer
        .collect()
 }

-fn resync_status_for_state(
-    state: &mut SiteReplicationState,
-    op_type: &str,
-    peer: &PeerInfo,
-    bucket_names: Vec<String>,
-) -> SRResyncOpStatus {
-    let status = SRResyncOpStatus {
-        op_type: op_type.to_string(),
-        resync_id: Uuid::new_v4().to_string(),
-        status: "success".to_string(),
-        buckets: bucket_names
-            .into_iter()
-            .map(|bucket| ResyncBucketStatus {
-                bucket,
-                status: if op_type == SITE_REPL_RESYNC_CANCEL {
-                    "canceled".to_string()
-                } else {
-                    "started".to_string()
-                },
-                ..Default::default()
-            })
-            .collect(),
-        ..Default::default()
+#[derive(Debug, Serialize, Deserialize)]
+struct SiteResyncContinuationToken {
+    id: String,
+    generation: u64,
+    offset: usize,
+}
+
+fn site_resync_is_active(status: &SRResyncOpStatus) -> bool {
+    matches!(status.state.as_str(), "pending" | "running" | "canceling")
+}
+
+fn site_resync_cancel_is_idempotent(status: &SRResyncOpStatus) -> bool {
+    status.state == "canceled"
+}
+
+fn site_resync_nonnegative(value: i64) -> u64 {
+    u64::try_from(value.max(0)).unwrap_or_default()
+}
+
+fn site_resync_bucket_state(status: replication::ResyncStatusType) -> &'static str {
+    match status {
+        replication::ResyncStatusType::ResyncPending => "pending",
+        replication::ResyncStatusType::ResyncStarted => "running",
+        replication::ResyncStatusType::ResyncCompleted => "completed",
+        replication::ResyncStatusType::ResyncCanceled => "canceled",
+        replication::ResyncStatusType::ResyncFailed | replication::ResyncStatusType::NoResync => "failed",
+    }
+}
+
+fn site_bucket_resync_is_active(status: replication::ResyncStatusType) -> bool {
+    matches!(
+        status,
+        replication::ResyncStatusType::ResyncPending | replication::ResyncStatusType::ResyncStarted
+    )
+}
+
+fn apply_site_resync_target_status(bucket: &mut ResyncBucketStatus, target: &replication::TargetReplicationResyncStatus) {
+    bucket.status = site_resync_bucket_state(target.resync_status).to_string();
+    bucket.started_at = target.start_time;
+    bucket.updated_at = target.last_update;
+    bucket.replicated_objects = site_resync_nonnegative(target.replicated_count);
+    bucket.replicated_bytes = site_resync_nonnegative(target.replicated_size);
+    bucket.failed_objects = site_resync_nonnegative(target.failed_count);
+    bucket.failed_bytes = site_resync_nonnegative(target.failed_size);
+    bucket.err_detail = target.error.as_deref().map(summarize_peer_error_detail).unwrap_or_default();
+    if matches!(bucket.status.as_str(), "completed" | "canceled" | "failed") {
+        bucket.completed_at = bucket.updated_at;
+    }
+}
+
+fn summarize_site_resync_status(status: &mut SRResyncOpStatus, now: OffsetDateTime) {
+    status.total_buckets = status.buckets.len() as u64;
+    status.pending_buckets = 0;
+    status.running_buckets = 0;
+    status.completed_buckets = 0;
+    status.failed_buckets = 0;
+    status.canceled_buckets = 0;
+    status.replicated_objects = 0;
+    status.replicated_bytes = 0;
+    status.failed_objects = 0;
+    status.failed_bytes = 0;
+
+    for bucket in &status.buckets {
+        match bucket.status.as_str() {
+            "pending" => status.pending_buckets += 1,
+            "running" | "started" => status.running_buckets += 1,
+            "completed" | "success" => status.completed_buckets += 1,
+            "canceled" => status.canceled_buckets += 1,
+            _ => status.failed_buckets += 1,
+        }
+        status.replicated_objects = status.replicated_objects.saturating_add(bucket.replicated_objects);
+        status.replicated_bytes = status.replicated_bytes.saturating_add(bucket.replicated_bytes);
+        status.failed_objects = status.failed_objects.saturating_add(bucket.failed_objects);
+        status.failed_bytes = status.failed_bytes.saturating_add(bucket.failed_bytes);
+    }
+
+    status.updated_at = Some(now);
+    status.status = if status.failed_buckets > 0 { "failed" } else { "success" }.to_string();
+    let has_active_buckets = status.pending_buckets > 0
+        || status.running_buckets > 0
+        || status.buckets.iter().any(|bucket| bucket.status == "conflict");
+    status.state = if has_active_buckets {
+        if status.op_type == SITE_REPL_RESYNC_CANCEL {
+            "canceling"
+        } else if status.running_buckets > 0 {
+            "running"
+        } else {
+            "pending"
+        }
+    } else if status.failed_buckets > 0 {
+        "failed"
+    } else if status.op_type == SITE_REPL_RESYNC_CANCEL || status.canceled_buckets == status.total_buckets {
+        "canceled"
+    } else {
+        "completed"
+    }
+    .to_string();
+    if matches!(status.state.as_str(), "completed" | "canceled" | "failed") && status.completed_at.is_none() {
+        status.completed_at = Some(now);
+    }
+    status.err_detail = if status.failed_buckets > 0 {
+        format!("{} of {} buckets failed", status.failed_buckets, status.total_buckets)
+    } else {
+        String::new()
    };
-    state.resync_status.insert(peer.deployment_id.clone(), status.clone());
-    status
+}
+
+fn site_resync_page(status: &SRResyncOpStatus, limit: usize, offset: usize) -> S3Result<SRResyncOpStatus> {
+    if offset > status.buckets.len() {
+        return Err(s3_error!(InvalidRequest, "invalid resync continuation token"));
+    }
+    let mut response = status.clone();
+    let end = offset.saturating_add(limit).min(status.buckets.len());
+    response.buckets = status.buckets[offset..end].to_vec();
+    response.truncated = end < status.buckets.len();
+    response.next_continuation_token = if response.truncated {
+        let token = SiteResyncContinuationToken {
+            id: status.resync_id.clone(),
+            generation: status.generation,
+            offset: end,
+        };
+        let encoded = serde_json::to_vec(&token)
+            .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("encode resync cursor failed: {err}")))?;
+        URL_SAFE_NO_PAD.encode(encoded)
+    } else {
+        String::new()
+    };
+    Ok(response)
+}
+
+fn parse_site_resync_page(query: &HashMap<String, String>, status: &SRResyncOpStatus) -> S3Result<(usize, usize)> {
+    let limit = query
+        .get("limit")
+        .map(|value| value.parse::<usize>())
+        .transpose()
+        .map_err(|_| s3_error!(InvalidRequest, "invalid resync page limit"))?
+        .unwrap_or(SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE);
+    if limit == 0 || limit > SITE_REPL_RESYNC_MAX_PAGE_SIZE {
+        return Err(s3_error!(InvalidRequest, "invalid resync page limit"));
+    }
+    let offset = if let Some(value) = query.get("continuationToken") {
+        let decoded = URL_SAFE_NO_PAD
+            .decode(value)
+            .map_err(|_| s3_error!(InvalidRequest, "invalid resync continuation token"))?;
+        let token: SiteResyncContinuationToken =
+            serde_json::from_slice(&decoded).map_err(|_| s3_error!(InvalidRequest, "invalid resync continuation token"))?;
+        if token.id != status.resync_id || token.generation != status.generation {
+            return Err(s3_error!(InvalidRequest, "stale resync continuation token"));
+        }
+        token.offset
+    } else {
+        0
+    };
+    Ok((limit, offset))
 }

 fn bucket_target_endpoint(target: &BucketTarget) -> String {
@@ -4924,8 +5058,10 @@ fn bucket_target_endpoint(target: &BucketTarget) -> String {
 }

 fn bucket_target_matches_peer(target: &BucketTarget, peer: &PeerInfo) -> bool {
-    (!target.deployment_id.is_empty() && target.deployment_id == peer.deployment_id)
-        || bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint)
+    if !target.deployment_id.is_empty() {
+        return target.deployment_id == peer.deployment_id;
+    }
+    bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint)
 }

 fn site_replication_target_arns_by_peer(config: Option<&s3s::dto::ReplicationConfiguration>) -> HashMap<String, String> {
@@ -5576,7 +5712,12 @@ async fn backfill_existing_buckets_after_add(
            if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
                continue;
            }
-            let result = start_site_bucket_resync(name, peer, &resync_id).await;
+            let manifest = site_bucket_resync_manifest_entry(name, peer, OffsetDateTime::now_utc()).await;
+            let result = if manifest.target_arn.is_empty() {
+                manifest
+            } else {
+                start_site_bucket_resync(name, &manifest.target_arn, &resync_id).await
+            };
            if result.status == "failed" {
                warn!(
                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
@@ -5657,10 +5798,60 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac
    Ok(())
 }

-async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str) -> ResyncBucketStatus {
+async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: OffsetDateTime) -> ResyncBucketStatus {
+    let mut entry = ResyncBucketStatus {
+        bucket: bucket.to_string(),
+        status: "pending".to_string(),
+        created_at: Some(now),
+        updated_at: Some(now),
+        ..Default::default()
+    };
+    let _targets_guard = lock_bucket_targets_metadata(bucket).await;
+    let (config, _) = match metadata_sys::get_replication_config(bucket).await {
+        Ok(config) => config,
+        Err(err) => {
+            entry.status = "failed".to_string();
+            entry.err_detail = summarize_peer_error_detail(&err.to_string());
+            return entry;
+        }
+    };
+    let targets = match metadata_sys::list_bucket_targets(bucket).await {
+        Ok(targets) => targets,
+        Err(err) => {
+            entry.status = "failed".to_string();
+            entry.err_detail = summarize_peer_error_detail(&err.to_string());
+            return entry;
+        }
+    };
+    let mut matching = targets
+        .targets
+        .iter()
+        .filter(|target| target.target_type == BucketTargetType::ReplicationService && bucket_target_matches_peer(target, peer));
+    let Some(target) = matching.next() else {
+        entry.status = "failed".to_string();
+        entry.err_detail = "no valid remote target found for peer".to_string();
+        return entry;
+    };
+    if matching.next().is_some() {
+        entry.status = "failed".to_string();
+        entry.err_detail = "multiple remote targets matched peer".to_string();
+        return entry;
+    }
+    let (has_arn, existing_object_enabled) = config.has_existing_object_replication(&target.arn);
+    if !has_arn || !existing_object_enabled {
+        entry.status = "failed".to_string();
+        entry.err_detail = "existing object replication is not enabled for the peer target".to_string();
+        return entry;
+    }
+    entry.target_arn = target.arn.clone();
+    entry
+}
+
+async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus {
    let mut bucket_status = ResyncBucketStatus {
        bucket: bucket.to_string(),
-        status: "started".to_string(),
+        target_arn: target_arn.to_string(),
+        status: "running".to_string(),
        ..Default::default()
    };
    let targets_guard = lock_bucket_targets_metadata(bucket).await;
@@ -5683,15 +5874,40 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str
        }
    };

+    let Some(pool) = current_replication_pool_handle() else {
+        bucket_status.status = "failed".to_string();
+        bucket_status.err_detail = "replication pool is not initialized".to_string();
+        return bucket_status;
+    };
+
+    let Some(target_index) = targets
+        .targets
+        .iter()
+        .position(|target| target.target_type == BucketTargetType::ReplicationService && target.arn == target_arn)
+    else {
+        bucket_status.status = "failed".to_string();
+        bucket_status.err_detail = "recorded remote target no longer exists".to_string();
+        return bucket_status;
+    };
+
+    let existing_reset_id = targets.targets[target_index].reset_id.clone();
+    if !existing_reset_id.is_empty() && existing_reset_id != resync_id {
+        let existing_is_active = pool
+            .get_bucket_resync_status(bucket)
+            .await
+            .ok()
+            .and_then(|status| status.targets_map.get(target_arn).cloned())
+            .is_none_or(|target| target.resync_id != existing_reset_id || site_bucket_resync_is_active(target.resync_status));
+        if existing_is_active {
+            bucket_status.status = "conflict".to_string();
+            bucket_status.err_detail = "target belongs to a different active resync operation".to_string();
+            return bucket_status;
+        }
+    }
+
    let reset_before = Some(OffsetDateTime::now_utc());
    let target_arn = {
-        let Some(target) = targets.targets.iter_mut().find(|target| {
-            target.target_type == BucketTargetType::ReplicationService && bucket_target_matches_peer(target, peer)
-        }) else {
-            bucket_status.status = "failed".to_string();
-            bucket_status.err_detail = format!("no valid remote target found for peer {}", peer.deployment_id);
-            return bucket_status;
-        };
+        let target = &mut targets.targets[target_index];

        let (has_arn, existing_object_enabled) = config.has_existing_object_replication(&target.arn);
        if !has_arn || !existing_object_enabled {
@@ -5722,12 +5938,6 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str
    BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await;
    drop(targets_guard);

-    let Some(pool) = current_replication_pool_handle() else {
-        bucket_status.status = "failed".to_string();
-        bucket_status.err_detail = "replication pool is not initialized".to_string();
-        return bucket_status;
-    };
-
    if let Err(err) = pool
        .start_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, reset_before))
        .await
@@ -5739,9 +5949,10 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str
    bucket_status
 }

-async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str) -> ResyncBucketStatus {
+async fn cancel_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus {
    let mut bucket_status = ResyncBucketStatus {
        bucket: bucket.to_string(),
+        target_arn: target_arn.to_string(),
        status: "canceled".to_string(),
        ..Default::default()
    };
@@ -5757,18 +5968,32 @@ async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &st
    };

    let Some(target) = targets.targets.iter_mut().find(|target| {
-        target.target_type == BucketTargetType::ReplicationService
-            && bucket_target_matches_peer(target, peer)
-            && target.reset_id == resync_id
+        target.target_type == BucketTargetType::ReplicationService && target.arn == target_arn && target.reset_id == resync_id
    }) else {
        bucket_status.status = "failed".to_string();
-        bucket_status.err_detail = format!("no in-progress resync target found for peer {}", peer.deployment_id);
+        bucket_status.err_detail = "recorded resync target is not in progress".to_string();
        return bucket_status;
    };

+    let target_arn = target.arn.clone();
+
+    let Some(pool) = current_replication_pool_handle() else {
+        bucket_status.status = "failed".to_string();
+        bucket_status.err_detail = "replication pool is not initialized".to_string();
+        return bucket_status;
+    };
+
+    if let Err(err) = pool
+        .cancel_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, None))
+        .await
+    {
+        bucket_status.status = "failed".to_string();
+        bucket_status.err_detail = err.to_string();
+        return bucket_status;
+    }
+
    target.reset_id.clear();
    target.reset_before_date = None;
-    let target_arn = target.arn.clone();

    let json_targets = match serde_json::to_vec(&targets) {
        Ok(json_targets) => json_targets,
@@ -5787,23 +6012,91 @@ async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &st
    BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await;
    drop(targets_guard);

-    let Some(pool) = current_replication_pool_handle() else {
-        bucket_status.status = "failed".to_string();
-        bucket_status.err_detail = "replication pool is not initialized".to_string();
-        return bucket_status;
-    };
-
-    if let Err(err) = pool
-        .cancel_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, None))
-        .await
-    {
-        bucket_status.status = "failed".to_string();
-        bucket_status.err_detail = err.to_string();
-    }
-
    bucket_status
 }

+async fn refresh_site_resync_status(mut status: SRResyncOpStatus, peer: &PeerInfo) -> SRResyncOpStatus {
+    for bucket in &mut status.buckets {
+        if bucket.target_arn.is_empty() && matches!(bucket.status.as_str(), "pending" | "running" | "started") {
+            let resolved = site_bucket_resync_manifest_entry(&bucket.bucket, peer, OffsetDateTime::now_utc()).await;
+            if resolved.target_arn.is_empty() {
+                bucket.status = "failed".to_string();
+                bucket.err_detail = resolved.err_detail;
+            } else {
+                bucket.target_arn = resolved.target_arn;
+                bucket.status = "pending".to_string();
+            }
+        }
+    }
+    if let Some(pool) = current_replication_pool_handle() {
+        for bucket in &mut status.buckets {
+            if bucket.target_arn.is_empty() || bucket.status == "failed" {
+                continue;
+            }
+            match pool.get_bucket_resync_status(&bucket.bucket).await {
+                Ok(live) => match live.targets_map.get(&bucket.target_arn) {
+                    Some(target) if target.resync_id == status.resync_id => {
+                        apply_site_resync_target_status(bucket, target);
+                    }
+                    Some(target) if !target.resync_id.is_empty() && site_bucket_resync_is_active(target.resync_status) => {
+                        bucket.status = "conflict".to_string();
+                        bucket.err_detail = "recorded target belongs to a different resync operation".to_string();
+                        bucket.updated_at = Some(OffsetDateTime::now_utc());
+                    }
+                    Some(target) if !target.resync_id.is_empty() => {
+                        bucket.status = "failed".to_string();
+                        bucket.err_detail = "recorded resync operation was superseded by a terminal bucket resync".to_string();
+                        bucket.updated_at = Some(OffsetDateTime::now_utc());
+                        bucket.completed_at = bucket.updated_at;
+                    }
+                    _ if matches!(bucket.status.as_str(), "pending" | "running" | "started") => {
+                        let previous = bucket.clone();
+                        let mut recovered =
+                            start_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await;
+                        recovered.created_at = previous.created_at;
+                        recovered.started_at = previous.started_at.or(Some(OffsetDateTime::now_utc()));
+                        recovered.updated_at = Some(OffsetDateTime::now_utc());
+                        recovered.generation = status.generation;
+                        recovered.err_detail = summarize_peer_error_detail(&recovered.err_detail);
+                        *bucket = recovered;
+                    }
+                    _ => {}
+                },
+                Err(err) => {
+                    bucket.err_detail = summarize_peer_error_detail(&err.to_string());
+                    bucket.updated_at = Some(OffsetDateTime::now_utc());
+                }
+            }
+        }
+    }
+    summarize_site_resync_status(&mut status, OffsetDateTime::now_utc());
+    status
+}
+
+async fn persist_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> {
+    let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await;
+    let mut state = load_site_replication_state().await?;
+    if state
+        .resync_status
+        .get(peer_id)
+        .is_some_and(|current| current.resync_id != status.resync_id || current.generation != status.generation)
+    {
+        return Err(s3_error!(InvalidRequest, "site replication resync state changed"));
+    }
+    state.resync_status.insert(peer_id.to_string(), status.clone());
+    save_site_replication_state(&state).await
+}
+
+async fn persist_new_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> {
+    let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await;
+    let mut state = load_site_replication_state().await?;
+    if state.resync_status.get(peer_id).is_some_and(site_resync_is_active) {
+        return Err(s3_error!(InvalidRequest, "site replication resync is already active"));
+    }
+    state.resync_status.insert(peer_id.to_string(), status.clone());
+    save_site_replication_state(&state).await
+}
+
 fn apply_state_edit_req(mut state: SiteReplicationState, body: SRStateEditReq) -> SiteReplicationState {
    let Some(incoming_updated_at) = body.updated_at else {
        return state;
@@ -7261,89 +7554,162 @@ pub struct SiteReplicationResyncOpHandler {}
 impl Operation for SiteReplicationResyncOpHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        validate_site_replication_admin_request(&req, AdminAction::SiteReplicationResyncAction).await?;
-        let operation = query_pairs(&req.uri).get("operation").cloned().unwrap_or_default();
-        // Resolve before the request body is consumed below; the `else` stays
-        // at its original position so error precedence is unchanged.
+        let query = query_pairs(&req.uri);
+        let operation = query.get("operation").cloned().unwrap_or_default();
        let resolved_store = object_store_from_req(&req);
-        let peer: PeerInfo = read_site_replication_json(req, "", false).await?;
-        let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await;
-        let mut state = load_site_replication_state().await?;
-        let local_peer = current_local_runtime_peer(&state);
-        let peer = normalize_peer_info(peer);
-        if peer.deployment_id == local_peer.deployment_id {
-            return Err(s3_error!(InvalidRequest, "invalid peer specified - cannot resync to self"));
-        }
-        if !state.peers.contains_key(&peer.deployment_id) {
-            return Err(s3_error!(InvalidRequest, "site replication peer not found"));
-        }
-        let Some(store) = resolved_store else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        let requested_peer: PeerInfo = read_site_replication_json(req, "", false).await?;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let (peer, existing_status) = {
+            let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await;
+            let state = load_site_replication_state().await?;
+            let local_peer = current_local_runtime_peer(&state);
+            let requested_peer = normalize_peer_info(requested_peer);
+            if requested_peer.deployment_id == local_peer.deployment_id {
+                return Err(s3_error!(InvalidRequest, "invalid peer specified - cannot resync to self"));
+            }
+            let peer = state
+                .peers
+                .get(&requested_peer.deployment_id)
+                .cloned()
+                .ok_or_else(|| s3_error!(InvalidRequest, "site replication peer not found"))?;
+            (peer, state.resync_status.get(&requested_peer.deployment_id).cloned())
        };
-        let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?;
-        let bucket_names: Vec<String> = buckets.into_iter().map(|bucket| bucket.name).collect();

-        let status = match operation.as_str() {
+        let mut status = match operation.as_str() {
            SITE_REPL_RESYNC_START => {
-                let mut status = resync_status_for_state(&mut state, &operation, &peer, vec![]);
-                let mut bucket_statuses = Vec::new();
+                if let Some(existing) = existing_status.as_ref() {
+                    let existing = refresh_site_resync_status(existing.clone(), &peer).await;
+                    persist_site_resync_status(&peer.deployment_id, &existing).await?;
+                    if site_resync_is_active(&existing) {
+                        return Err(s3_error!(InvalidRequest, "site replication resync is already active"));
+                    }
+                }
+                let Some(store) = resolved_store else {
+                    return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+                };
+                let mut bucket_names: Vec<String> = store
+                    .list_bucket(&BucketOptions::default())
+                    .await
+                    .map_err(ApiError::from)?
+                    .into_iter()
+                    .map(|bucket| bucket.name)
+                    .collect();
+                bucket_names.sort();
+                let now = OffsetDateTime::now_utc();
+                let mut bucket_statuses = Vec::with_capacity(bucket_names.len());
                for bucket in bucket_names {
-                    bucket_statuses.push(start_site_bucket_resync(&bucket, &peer, &status.resync_id).await);
+                    bucket_statuses.push(site_bucket_resync_manifest_entry(&bucket, &peer, now).await);
                }
-                let failures = bucket_statuses.iter().filter(|bucket| bucket.status == "failed").count();
-                if failures == bucket_statuses.len() && !bucket_statuses.is_empty() {
-                    status.status = "failed".to_string();
-                    status.err_detail = "all buckets resync failed".to_string();
-                } else if failures > 0 {
-                    status.err_detail = "partial failure in starting site resync".to_string();
+                let mut status = SRResyncOpStatus {
+                    op_type: SITE_REPL_RESYNC_START.to_string(),
+                    resync_id: Uuid::new_v4().to_string(),
+                    status: "success".to_string(),
+                    state: "pending".to_string(),
+                    buckets: bucket_statuses,
+                    created_at: Some(now),
+                    started_at: Some(now),
+                    updated_at: Some(now),
+                    generation: existing_status
+                        .as_ref()
+                        .map_or(1, |existing| existing.generation.saturating_add(1).max(1)),
+                    ..Default::default()
+                };
+                summarize_site_resync_status(&mut status, now);
+                persist_new_site_resync_status(&peer.deployment_id, &status).await?;
+                for index in 0..status.buckets.len() {
+                    if status.buckets[index].target_arn.is_empty() || status.buckets[index].status == "failed" {
+                        continue;
+                    }
+                    let previous = status.buckets[index].clone();
+                    let mut result = start_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await;
+                    result.created_at = previous.created_at;
+                    result.started_at = Some(OffsetDateTime::now_utc());
+                    result.updated_at = result.started_at;
+                    result.generation = status.generation;
+                    result.err_detail = summarize_peer_error_detail(&result.err_detail);
+                    status.buckets[index] = result;
+                    summarize_site_resync_status(&mut status, OffsetDateTime::now_utc());
+                    persist_site_resync_status(&peer.deployment_id, &status).await?;
                }
-                status.buckets = bucket_statuses;
-                state.resync_status.insert(peer.deployment_id.clone(), status.clone());
+                status = refresh_site_resync_status(status, &peer).await;
+                persist_site_resync_status(&peer.deployment_id, &status).await?;
                status
            }
            SITE_REPL_RESYNC_CANCEL => {
-                let Some(existing_status) = state.resync_status.get(&peer.deployment_id).cloned() else {
+                let Some(existing_status) = existing_status else {
                    return Err(s3_error!(InvalidRequest, "no resync in progress"));
                };
                if existing_status.resync_id.is_empty() {
                    return Err(s3_error!(InvalidRequest, "no resync in progress"));
                }
-                let mut status = SRResyncOpStatus {
-                    op_type: operation.clone(),
-                    resync_id: existing_status.resync_id.clone(),
-                    status: "success".to_string(),
-                    ..Default::default()
-                };
-                let mut bucket_statuses = Vec::new();
-                for bucket in bucket_names {
-                    bucket_statuses.push(cancel_site_bucket_resync(&bucket, &peer, &existing_status.resync_id).await);
+                let mut status = refresh_site_resync_status(existing_status, &peer).await;
+                if status.buckets.iter().any(|bucket| bucket.status == "conflict") {
+                    return Err(s3_error!(
+                        InvalidRequest,
+                        "site replication resync target belongs to a different active operation"
+                    ));
                }
-                let failures = bucket_statuses.iter().filter(|bucket| bucket.status == "failed").count();
-                if failures == bucket_statuses.len() && !bucket_statuses.is_empty() {
-                    status.status = "failed".to_string();
-                    status.err_detail = "all buckets resync cancel failed".to_string();
-                } else if failures > 0 {
-                    status.err_detail = "partial failure in canceling site resync".to_string();
+                if site_resync_cancel_is_idempotent(&status) {
+                    status.op_type = SITE_REPL_RESYNC_CANCEL.to_string();
+                    for bucket in &status.buckets {
+                        if !bucket.target_arn.is_empty() {
+                            let _ = cancel_site_bucket_resync(&bucket.bucket, &bucket.target_arn, &status.resync_id).await;
+                        }
+                    }
+                    status
+                } else {
+                    if !site_resync_is_active(&status) {
+                        return Err(s3_error!(InvalidRequest, "no active resync to cancel"));
+                    }
+                    status.op_type = SITE_REPL_RESYNC_CANCEL.to_string();
+                    status.state = "canceling".to_string();
+                    status.updated_at = Some(OffsetDateTime::now_utc());
+                    persist_site_resync_status(&peer.deployment_id, &status).await?;
+                    for index in 0..status.buckets.len() {
+                        if status.buckets[index].target_arn.is_empty()
+                            || matches!(status.buckets[index].status.as_str(), "failed" | "canceled")
+                        {
+                            continue;
+                        }
+                        let previous = status.buckets[index].clone();
+                        let mut result =
+                            cancel_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await;
+                        result.created_at = previous.created_at;
+                        result.started_at = previous.started_at;
+                        result.updated_at = Some(OffsetDateTime::now_utc());
+                        result.completed_at = result.updated_at;
+                        result.generation = status.generation;
+                        result.err_detail = summarize_peer_error_detail(&result.err_detail);
+                        status.buckets[index] = result;
+                        summarize_site_resync_status(&mut status, OffsetDateTime::now_utc());
+                        persist_site_resync_status(&peer.deployment_id, &status).await?;
+                    }
+                    status = refresh_site_resync_status(status, &peer).await;
+                    persist_site_resync_status(&peer.deployment_id, &status).await?;
+                    status
                }
-                status.buckets = bucket_statuses;
-                state.resync_status.insert(peer.deployment_id.clone(), status.clone());
-                status
            }
            SITE_REPL_RESYNC_STATUS => {
-                let status = state
-                    .resync_status
-                    .get(&peer.deployment_id)
-                    .cloned()
-                    .unwrap_or_else(|| SRResyncOpStatus {
-                        op_type: SITE_REPL_RESYNC_STATUS.to_string(),
-                        status: "not-found".to_string(),
-                        ..Default::default()
-                    });
-                return json_response(&status);
+                let status = existing_status.unwrap_or_else(|| SRResyncOpStatus {
+                    op_type: SITE_REPL_RESYNC_STATUS.to_string(),
+                    status: "not-found".to_string(),
+                    ..Default::default()
+                });
+                if status.resync_id.is_empty() {
+                    status
+                } else {
+                    let status = refresh_site_resync_status(status, &peer).await;
+                    persist_site_resync_status(&peer.deployment_id, &status).await?;
+                    status
+                }
            }
            _ => return Err(s3_error!(InvalidRequest, "unsupported resync operation")),
        };
-        save_site_replication_state(&state).await?;
-        json_response(&status)
+        status
+            .buckets
+            .sort_by(|left, right| left.bucket.cmp(&right.bucket).then(left.target_arn.cmp(&right.target_arn)));
+        let (limit, offset) = parse_site_resync_page(&query, &status)?;
+        json_response(&site_resync_page(&status, limit, offset)?)
    }
 }

@@ -8860,6 +9226,57 @@ mod tests {
        assert!(!policy.is_allowed(&put_policy_args).await);
    }

+    // The replication service account must be able to carry object-lock metadata to the peer.
+    // Without these actions the peer answers AccessDenied for any replicated object that has
+    // retention or a legal hold, so a WORM-protected object never reaches the replica at all,
+    // and a retention change made after upload never propagates.
+    #[tokio::test]
+    async fn test_site_replicator_policy_allows_object_lock_replication() {
+        let policy = site_replicator_service_account_policy().expect("site replicator policy should parse");
+        let groups: Option<Vec<String>> = None;
+        let claims = HashMap::new();
+        let conditions = HashMap::new();
+
+        let base_args = rustfs_policy::policy::Args {
+            account: SITE_REPLICATOR_SERVICE_ACCOUNT,
+            groups: &groups,
+            action: Action::S3Action(S3Action::PutObjectRetentionAction),
+            conditions: &conditions,
+            is_owner: false,
+            claims: &claims,
+            deny_only: false,
+            bucket: "photos",
+            object: "image.jpg",
+        };
+
+        for action in [
+            S3Action::PutObjectRetentionAction,
+            S3Action::GetObjectRetentionAction,
+            S3Action::PutObjectLegalHoldAction,
+            S3Action::GetObjectLegalHoldAction,
+        ] {
+            let args = rustfs_policy::policy::Args {
+                action: Action::S3Action(action),
+                ..base_args
+            };
+            assert!(
+                policy.is_allowed(&args).await,
+                "site replicator must be allowed to replicate object-lock metadata: {action:?}"
+            );
+        }
+
+        // Governance bypass stays denied: replication must not be able to erase a retained
+        // version on the peer.
+        let bypass_args = rustfs_policy::policy::Args {
+            action: Action::S3Action(S3Action::BypassGovernanceRetentionAction),
+            ..base_args
+        };
+        assert!(
+            !policy.is_allowed(&bypass_args).await,
+            "site replicator must not be granted governance bypass"
+        );
+    }
+
    #[test]
    fn test_sr_peer_edit_handler_uses_site_replication_operation_action() {
        let src = include_str!("site_replication.rs");
@@ -11485,4 +11902,88 @@ mod tests {
        assert!(rule_ids.contains(&"site-repl-dep-b"));
        assert!(rule_ids.contains(&"site-repl-dep-c"));
    }
+
+    #[test]
+    fn site_resync_summary_reports_partial_failure_and_clamps_counters() {
+        let now = OffsetDateTime::now_utc();
+        let mut running = ResyncBucketStatus {
+            bucket: "b".to_string(),
+            target_arn: "arn-b".to_string(),
+            ..Default::default()
+        };
+        apply_site_resync_target_status(
+            &mut running,
+            &replication::TargetReplicationResyncStatus {
+                resync_status: replication::ResyncStatusType::ResyncStarted,
+                resync_id: "run-1".to_string(),
+                replicated_count: 4,
+                replicated_size: 16,
+                failed_count: -1,
+                failed_size: -2,
+                ..Default::default()
+            },
+        );
+        let failed = ResyncBucketStatus {
+            bucket: "a".to_string(),
+            target_arn: "arn-a".to_string(),
+            status: "conflict".to_string(),
+            err_detail: "durable failure".to_string(),
+            ..Default::default()
+        };
+        let mut status = SRResyncOpStatus {
+            op_type: SITE_REPL_RESYNC_START.to_string(),
+            resync_id: "run-1".to_string(),
+            buckets: vec![running, failed],
+            ..Default::default()
+        };
+
+        summarize_site_resync_status(&mut status, now);
+
+        assert_eq!(status.status, "failed");
+        assert_eq!(status.state, "running");
+        assert_eq!(status.running_buckets, 1);
+        assert_eq!(status.failed_buckets, 1);
+        assert_eq!(status.replicated_objects, 4);
+        assert_eq!(status.replicated_bytes, 16);
+        assert_eq!(status.failed_objects, 0);
+        assert_eq!(status.failed_bytes, 0);
+        assert_eq!(status.completed_at, None);
+        assert!(site_resync_is_active(&status));
+        assert!(site_resync_cancel_is_idempotent(&SRResyncOpStatus {
+            state: "canceled".to_string(),
+            ..Default::default()
+        }));
+        assert!(site_bucket_resync_is_active(replication::ResyncStatusType::ResyncPending));
+        assert!(site_bucket_resync_is_active(replication::ResyncStatusType::ResyncStarted));
+        assert!(!site_bucket_resync_is_active(replication::ResyncStatusType::ResyncCompleted));
+    }
+
+    #[test]
+    fn site_resync_pagination_is_sorted_and_rejects_stale_cursor() {
+        let status = SRResyncOpStatus {
+            resync_id: "run-1".to_string(),
+            generation: 3,
+            buckets: ["a", "b", "c"]
+                .into_iter()
+                .map(|bucket| ResyncBucketStatus {
+                    bucket: bucket.to_string(),
+                    ..Default::default()
+                })
+                .collect(),
+            ..Default::default()
+        };
+        let first = site_resync_page(&status, 2, 0).expect("first page should be valid");
+        assert!(first.truncated);
+        assert_eq!(first.buckets.iter().map(|bucket| bucket.bucket.as_str()).collect::<Vec<_>>(), ["a", "b"]);
+
+        let query = HashMap::from([
+            ("limit".to_string(), "2".to_string()),
+            ("continuationToken".to_string(), first.next_continuation_token),
+        ]);
+        let (_, offset) = parse_site_resync_page(&query, &status).expect("cursor should match operation");
+        assert_eq!(offset, 2);
+        let mut newer = status;
+        newer.generation += 1;
+        assert!(parse_site_resync_page(&query, &newer).is_err());
+    }
 }
@@ -14,6 +14,9 @@

 use super::{cluster_snapshot, metrics};
 use crate::admin::auth::validate_admin_request;
+use crate::admin::route_policy::{
+    ADMIN_ROUTE_POLICY_SPECS, DEFERRED_ADMIN_ROUTE_POLICIES, DeferredAdminRoutePolicy, DeferredRoutePolicyReason,
+};
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
    DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, object_store_from_req,
@@ -31,9 +34,10 @@ use matchit::Params;
 use rustfs_concurrency::WorkloadAdmissionRegistrySnapshot;
 use rustfs_madmin::{InfoMessage, StorageInfo};
 use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
+use rustfs_security_governance::{AdminRouteSpec, HttpMethod};
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use std::sync::atomic::{AtomicBool, Ordering};
 use tracing::{error, info, warn};

@@ -54,6 +58,9 @@ const OBSERVABILITY_SUMMARY_RESOLVED: &str = "observability summary resolved fro
 const TOPOLOGY_SUMMARY_RESOLVED: &str = "topology summary resolved from capability snapshot";
 const TOPOLOGY_SNAPSHOT_NOT_AVAILABLE: &str = "endpoint topology is not available before storage endpoint pools initialize";
 pub(crate) const RUNTIME_CAPABILITIES_ROUTE_SUFFIX: &str = "/v4/runtime/capabilities";
+const SITE_REPLICATION_INFO_ROUTE: &str = "/rustfs/admin/v3/site-replication/info";
+const SITE_REPLICATION_EDIT_ROUTE: &str = "/rustfs/admin/v3/site-replication/edit";
+const SITE_REPLICATION_RESYNC_ROUTE: &str = "/rustfs/admin/v3/site-replication/resync/op";

 macro_rules! log_system_request_rejected {
    ($operation:expr, $reason:expr) => {
@@ -609,7 +616,7 @@ impl Operation for StorageInfoHandler {

 pub struct DataUsageInfoHandler {}

-#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RuntimeCapabilitiesSummary {
    pub observability: CapabilityStatus,
    pub userspace_profiling: CapabilityStatus,
@@ -617,6 +624,12 @@ pub struct RuntimeCapabilitiesSummary {
    pub platform: CapabilityStatus,
    pub topology: CapabilityStatus,
    pub cluster_snapshot: CapabilityStatus,
+    #[serde(default)]
+    pub site_replication_info: CapabilityStatus,
+    #[serde(default)]
+    pub site_replication_edit: CapabilityStatus,
+    #[serde(default)]
+    pub site_replication_resync: CapabilityStatus,
 }

 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -729,9 +742,42 @@ fn build_runtime_capabilities_summary(
        cluster_snapshot: cluster_snapshot_summary.cloned().unwrap_or_else(|| {
            CapabilityStatus::unknown().with_reason("cluster snapshot is not available before storage endpoint pools initialize")
        }),
+        site_replication_info: admin_route_capability(HttpMethod::Get, SITE_REPLICATION_INFO_ROUTE),
+        site_replication_edit: admin_route_capability(HttpMethod::Put, SITE_REPLICATION_EDIT_ROUTE),
+        site_replication_resync: admin_route_capability(HttpMethod::Put, SITE_REPLICATION_RESYNC_ROUTE),
    }
 }

+fn admin_route_capability(method: HttpMethod, path: &str) -> CapabilityStatus {
+    admin_route_capability_from_inventory(method, path, ADMIN_ROUTE_POLICY_SPECS, DEFERRED_ADMIN_ROUTE_POLICIES)
+}
+
+fn admin_route_capability_from_inventory(
+    method: HttpMethod,
+    path: &str,
+    direct: &[AdminRouteSpec],
+    deferred: &[DeferredAdminRoutePolicy],
+) -> CapabilityStatus {
+    if direct.iter().any(|spec| spec.method() == method && spec.path() == path) {
+        return CapabilityStatus::supported().with_reason("public admin route is registered with an implemented handler");
+    }
+
+    match deferred
+        .iter()
+        .find(|policy| policy.method() == method && policy.path() == path)
+        .map(|policy| policy.reason())
+    {
+        Some(DeferredRoutePolicyReason::NotImplemented) | None => {
+            CapabilityStatus::unsupported().with_reason("public admin route is absent or not implemented")
+        }
+        Some(_) => CapabilityStatus::supported().with_reason("public admin route is registered with contextual authorization"),
+    }
+}
+
+fn runtime_capabilities_gate_actions() -> Vec<Action> {
+    vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]
+}
+
 fn summarize_named_capability_statuses<const N: usize>(
    statuses: [(&'static str, &CapabilityStatus); N],
    subject: &'static str,
@@ -767,15 +813,7 @@ impl Operation for RuntimeCapabilitiesHandler {
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
-        validate_admin_request(
-            &req.headers,
-            &cred,
-            owner,
-            false,
-            vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
-            remote_addr,
-        )
-        .await?;
+        validate_admin_request(&req.headers, &cred, owner, false, runtime_capabilities_gate_actions(), remote_addr).await?;

        let response = build_runtime_capabilities_response().await.map_err(|err| {
            log_system_request_failed!("runtime_capabilities", "build_runtime_capabilities_failed", err);
@@ -838,18 +876,26 @@ impl Operation for DataUsageInfoHandler {
 #[cfg(test)]
 mod tests {
    use super::{
-        OBSERVABILITY_SUMMARY_RESOLVED, ServerInfoResponse, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED,
-        build_runtime_capabilities_response, build_runtime_capabilities_summary, data_usage_info_gate_actions,
-        system_admin_discovery,
+        OBSERVABILITY_SUMMARY_RESOLVED, RuntimeCapabilitiesHandler, SITE_REPLICATION_EDIT_ROUTE, SITE_REPLICATION_INFO_ROUTE,
+        SITE_REPLICATION_RESYNC_ROUTE, ServerInfoResponse, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED,
+        admin_route_capability_from_inventory, build_runtime_capabilities_response, build_runtime_capabilities_summary,
+        data_usage_info_gate_actions, runtime_capabilities_gate_actions, system_admin_discovery,
    };
+    use crate::admin::router::Operation;
    use crate::admin::runtime_sources::DefaultAdminUsecase;
    use crate::admin::storage_api::cluster::{
        CapabilityState, CapabilityStatus, MemorySamplingState, ObservabilitySnapshot, PlatformSupport, TopologyCapabilities,
        TopologySnapshot, UserspaceProfilingCapability,
    };
+    use http::{Extensions, HeaderMap, Uri};
+    use hyper::Method;
+    use matchit::Params;
    use rustfs_concurrency::WorkloadClass;
    use rustfs_madmin::{InfoMessage, StorageInfo};
    use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
+    use rustfs_security_governance::HttpMethod;
+    use s3s::{Body, S3ErrorCode, S3Request};
+    use serde_json::json;

    /// Authz regression pin (rustfs/backlog#1306): datausageinfo stays an
    /// any-of gate over exactly DataUsageInfoAdminAction OR ListBucketAction.
@@ -882,6 +928,111 @@ mod tests {
        assert_eq!(response.summary.cluster_snapshot.state, CapabilityState::Unknown);
        assert_eq!(response.observability.platform.os.as_deref(), Some(std::env::consts::OS));
        assert_eq!(response.workload_admission.entries().len(), WorkloadClass::REQUIRED.len());
+        assert_eq!(response.summary.site_replication_info.state, CapabilityState::Supported);
+        assert_eq!(response.summary.site_replication_edit.state, CapabilityState::Supported);
+        assert_eq!(response.summary.site_replication_resync.state, CapabilityState::Supported);
+
+        let value = serde_json::to_value(response).expect("runtime capability response should serialize");
+        assert_eq!(value["summary"]["site_replication_info"]["state"], "supported");
+        assert_eq!(value["summary"]["site_replication_edit"]["state"], "supported");
+        assert_eq!(value["summary"]["site_replication_resync"]["state"], "supported");
+    }
+
+    #[test]
+    fn site_replication_capabilities_follow_public_route_inventory() {
+        for (method, path) in [
+            (HttpMethod::Get, SITE_REPLICATION_INFO_ROUTE),
+            (HttpMethod::Put, SITE_REPLICATION_EDIT_ROUTE),
+            (HttpMethod::Put, SITE_REPLICATION_RESYNC_ROUTE),
+        ] {
+            assert_eq!(
+                admin_route_capability_from_inventory(
+                    method,
+                    path,
+                    crate::admin::route_policy::ADMIN_ROUTE_POLICY_SPECS,
+                    crate::admin::route_policy::DEFERRED_ADMIN_ROUTE_POLICIES,
+                )
+                .state,
+                CapabilityState::Supported,
+            );
+        }
+    }
+
+    #[test]
+    fn site_replication_capabilities_reject_absent_internal_only_and_not_implemented_routes() {
+        use crate::admin::route_policy::{DeferredAdminRoutePolicy, DeferredRoutePolicyReason};
+
+        let not_implemented = [DeferredAdminRoutePolicy::new(
+            HttpMethod::Put,
+            SITE_REPLICATION_EDIT_ROUTE,
+            DeferredRoutePolicyReason::NotImplemented,
+        )];
+        assert_eq!(
+            admin_route_capability_from_inventory(HttpMethod::Put, SITE_REPLICATION_EDIT_ROUTE, &[], &not_implemented).state,
+            CapabilityState::Unsupported,
+        );
+        assert_eq!(
+            admin_route_capability_from_inventory(HttpMethod::Put, SITE_REPLICATION_EDIT_ROUTE, &[], &[]).state,
+            CapabilityState::Unsupported,
+        );
+
+        let internal_only = crate::admin::route_policy::ADMIN_ROUTE_POLICY_SPECS
+            .iter()
+            .copied()
+            .filter(|spec| spec.path() == "/rustfs/admin/v3/site-replication/peer/edit-capabilities")
+            .collect::<Vec<_>>();
+        assert_eq!(
+            admin_route_capability_from_inventory(HttpMethod::Put, SITE_REPLICATION_EDIT_ROUTE, &internal_only, &[]).state,
+            CapabilityState::Unsupported,
+        );
+    }
+
+    #[test]
+    fn older_runtime_summary_defaults_site_replication_capabilities_to_unknown() {
+        let summary: super::RuntimeCapabilitiesSummary = serde_json::from_value(json!({
+            "observability": { "state": "supported" },
+            "userspace_profiling": { "state": "supported" },
+            "memory_sampling": { "state": "supported" },
+            "platform": { "state": "supported" },
+            "topology": { "state": "supported" },
+            "cluster_snapshot": { "state": "supported" }
+        }))
+        .expect("older runtime summary should remain compatible");
+
+        assert_eq!(summary.site_replication_info.state, CapabilityState::Unknown);
+        assert_eq!(summary.site_replication_edit.state, CapabilityState::Unknown);
+        assert_eq!(summary.site_replication_resync.state, CapabilityState::Unknown);
+    }
+
+    #[tokio::test]
+    async fn runtime_capabilities_authentication_failure_is_not_reported_as_unsupported() {
+        let request = S3Request {
+            input: Body::empty(),
+            method: Method::GET,
+            uri: Uri::from_static("/rustfs/admin/v4/runtime/capabilities"),
+            headers: HeaderMap::new(),
+            extensions: Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        let error = RuntimeCapabilitiesHandler {}
+            .call(request, Params::new())
+            .await
+            .expect_err("runtime capabilities must reject unauthenticated requests before capability resolution");
+        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    /// Authorization denial for this exact action is pinned to AccessDenied by
+    /// `crate::admin::auth::tests::non_admin_credential_is_denied`.
+    #[test]
+    fn runtime_capabilities_gate_requires_server_info_authorization() {
+        assert_eq!(
+            runtime_capabilities_gate_actions(),
+            vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]
+        );
    }

    #[test]
@@ -65,7 +65,7 @@ use rustfs_notify::{Event as NotificationEvent, notification_system};
 use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_s3_types::EventName;
 use rustfs_signer::pre_sign_v4;
-use rustfs_utils::egress::validate_outbound_url;
+use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy};
 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,
@@ -229,7 +229,7 @@ struct ListenNotificationFilter {
    suffix: Option<String>,
 }

-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone)]
 struct ObjectLambdaWebhookConfig {
    endpoint: Url,
    auth_token: String,
@@ -238,6 +238,7 @@ struct ObjectLambdaWebhookConfig {
    client_ca: String,
    skip_tls_verify: bool,
    response_header_timeout: Option<Duration>,
+    outbound_resolver: OutboundDnsResolver,
 }

 const LAMBDA_WEBHOOK_SUB_SYS: &str = "lambda_webhook";
@@ -612,7 +613,8 @@ fn resolve_object_lambda_webhook_config_from_server_config(

    let parsed_endpoint =
        Url::parse(&endpoint).map_err(|_| s3_error!(InvalidRequest, "object lambda target endpoint is invalid"))?;
-    validate_outbound_url(&parsed_endpoint)
+    let outbound_resolver = outbound_policy()?
+        .resolver_for(&parsed_endpoint)
        .map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?;

    Ok(ObjectLambdaWebhookConfig {
@@ -623,6 +625,7 @@ fn resolve_object_lambda_webhook_config_from_server_config(
        client_ca: kvs.lookup(WEBHOOK_CLIENT_CA).unwrap_or_default(),
        skip_tls_verify: config_enable_is_on(&kvs.lookup(WEBHOOK_SKIP_TLS_VERIFY).unwrap_or_default()),
        response_header_timeout,
+        outbound_resolver,
    })
 }

@@ -658,9 +661,18 @@ async fn resolve_object_lambda_webhook_config(uri: &Uri) -> S3Result<ObjectLambd
 }

 fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Result<reqwest::Client> {
-    validate_outbound_url(&config.endpoint)
-        .map_err(|err| s3_error!(InvalidRequest, "object lambda target endpoint is not allowed: {}", err))?;
-    let mut builder = reqwest::Client::builder().user_agent(rustfs_targets::get_user_agent(rustfs_targets::ServiceType::Basis));
+    build_object_lambda_http_client_with_resolver(config, config.outbound_resolver.clone())
+}
+
+fn build_object_lambda_http_client_with_resolver(
+    config: &ObjectLambdaWebhookConfig,
+    resolver: impl reqwest::dns::Resolve + 'static,
+) -> S3Result<reqwest::Client> {
+    let mut builder = reqwest::Client::builder()
+        .no_proxy()
+        .dns_resolver(resolver)
+        .redirect(reqwest::redirect::Policy::none())
+        .user_agent(rustfs_targets::get_user_agent(rustfs_targets::ServiceType::Basis));

    if let Some(timeout) = config.response_header_timeout {
        builder = builder.timeout(timeout);
@@ -672,7 +684,7 @@ fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Resu
            component = LOG_COMPONENT_ADMIN,
            subsystem = LOG_SUBSYSTEM_OBJECT_LAMBDA,
            result = "tls_verification_disabled",
-            endpoint = %config.endpoint,
+            endpoint_origin = %config.endpoint.origin().ascii_serialization(),
            "admin router state"
        );
        builder = builder.danger_accept_invalid_certs(true);
@@ -706,6 +718,17 @@ fn build_object_lambda_http_client(config: &ObjectLambdaWebhookConfig) -> S3Resu
        .map_err(|e| s3_error!(InternalError, "failed to build object lambda http client: {e}"))
 }

+async fn send_object_lambda_request(request: reqwest::RequestBuilder) -> S3Result<reqwest::Response> {
+    request
+        .send()
+        .await
+        .map_err(|_| s3_error!(InternalError, "object lambda target request failed"))
+}
+
+fn outbound_policy() -> S3Result<&'static OutboundPolicy> {
+    OutboundPolicy::from_env_cached().map_err(|err| s3_error!(InvalidRequest, "invalid outbound policy: {}", err))
+}
+
 fn extract_request_scheme(headers: &HeaderMap, uri: &Uri) -> String {
    get_source_scheme(headers)
        .and_then(|value| {
@@ -1098,10 +1121,7 @@ async fn invoke_object_lambda_target(
        request_builder = request_builder.header("x-rustfs-object-lambda-version-id", version_id);
    }

-    let lambda_response = request_builder
-        .send()
-        .await
-        .map_err(|e| s3_error!(InternalError, "object lambda target request failed: {e}"))?;
+    let lambda_response = send_object_lambda_request(request_builder).await?;

    let status = lambda_response.status();
    let lambda_headers = lambda_response.headers().clone();
@@ -2576,7 +2596,70 @@ mod tests {
    use http::Method;
    use http::Uri;
    use s3s::S3Request;
+    use std::net::{IpAddr, SocketAddr};
    use time::macros::datetime;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    #[derive(Clone)]
+    struct StaticResolver(IpAddr);
+
+    impl reqwest::dns::Resolve for StaticResolver {
+        fn resolve(&self, _name: reqwest::dns::Name) -> reqwest::dns::Resolving {
+            let address = SocketAddr::new(self.0, 0);
+            Box::pin(async move { Ok(Box::new(std::iter::once(address)) as reqwest::dns::Addrs) })
+        }
+    }
+
+    fn object_lambda_test_config(endpoint: Url) -> ObjectLambdaWebhookConfig {
+        object_lambda_test_config_with_policy(endpoint, &OutboundPolicy::default())
+    }
+
+    fn object_lambda_test_config_with_policy(endpoint: Url, policy: &OutboundPolicy) -> ObjectLambdaWebhookConfig {
+        let outbound_resolver = policy
+            .resolver_for(&endpoint)
+            .expect("test endpoint should satisfy its outbound policy");
+        ObjectLambdaWebhookConfig {
+            endpoint,
+            auth_token: String::new(),
+            client_cert: String::new(),
+            client_key: String::new(),
+            client_ca: String::new(),
+            skip_tls_verify: false,
+            response_header_timeout: Some(Duration::from_secs(2)),
+            outbound_resolver,
+        }
+    }
+
+    #[derive(Clone, Default)]
+    struct CapturedLog(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    struct CapturedLogWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().expect("captured log lock").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLog {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'writer self) -> Self::Writer {
+            CapturedLogWriter(self.0.clone())
+        }
+    }
+
+    impl CapturedLog {
+        fn contents(&self) -> String {
+            String::from_utf8(self.0.lock().expect("captured log lock").clone()).expect("captured logs must be UTF-8")
+        }
+    }

    #[test]
    fn canonicalize_admin_path_maps_compat_prefix_to_rustfs_prefix() {
@@ -3778,6 +3861,214 @@ mod tests {
        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
    }

+    #[test]
+    fn object_lambda_client_ignores_proxies_and_does_not_follow_redirects() {
+        const CHILD_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_PROXY_CHILD";
+        const TARGET_URL_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_TARGET_URL";
+        const TARGET_ADDR_ENV: &str = "RUSTFS_TEST_OBJECT_LAMBDA_TARGET_ADDR";
+
+        if std::env::var_os(CHILD_ENV).is_some() {
+            let endpoint = Url::parse(&std::env::var(TARGET_URL_ENV).expect("child target URL")).expect("target URL");
+            let address = std::env::var(TARGET_ADDR_ENV)
+                .expect("child target address")
+                .parse::<SocketAddr>()
+                .expect("target address");
+            let config = object_lambda_test_config(endpoint);
+            let client = build_object_lambda_http_client_with_resolver(&config, StaticResolver(address.ip()))
+                .expect("object lambda client should build");
+            tokio::runtime::Runtime::new().expect("child runtime").block_on(async {
+                let response = client
+                    .get(config.endpoint)
+                    .send()
+                    .await
+                    .expect("object lambda request should bypass environment proxy");
+                assert_eq!(response.status(), reqwest::StatusCode::FOUND);
+            });
+            return;
+        }
+
+        use std::io::{Read, Write};
+        use std::net::TcpListener as StdTcpListener;
+
+        let target_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind object lambda target");
+        let target_address = target_listener.local_addr().expect("object lambda target address");
+        let redirect_listener = StdTcpListener::bind("127.0.0.1:0").expect("bind redirect destination");
+        let redirect_address = redirect_listener.local_addr().expect("redirect destination address");
+        drop(redirect_listener);
+        let target = std::thread::spawn(move || {
+            let (mut stream, _) = target_listener.accept().expect("accept object lambda request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).expect("read object lambda request");
+            assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /transform HTTP/1.1"));
+            stream
+                .write_all(
+                    format!(
+                        "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:{}/metadata\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
+                        redirect_address.port()
+                    )
+                    .as_bytes(),
+                )
+                .expect("write object lambda redirect");
+        });
+        let proxy_listener = StdTcpListener::bind("127.0.0.1:0").expect("reserve refused proxy address");
+        let proxy_address = proxy_listener.local_addr().expect("proxy address");
+        drop(proxy_listener);
+        let target_url = format!("http://object-lambda.test:{}/transform", target_address.port());
+        let proxy_url = format!("http://{proxy_address}");
+        let output = std::process::Command::new(std::env::current_exe().expect("resolve current test executable"))
+            .arg("object_lambda_client_ignores_proxies_and_does_not_follow_redirects")
+            .arg("--nocapture")
+            .env(CHILD_ENV, "1")
+            .env(TARGET_URL_ENV, target_url)
+            .env(TARGET_ADDR_ENV, target_address.to_string())
+            .env("HTTP_PROXY", &proxy_url)
+            .env("HTTPS_PROXY", &proxy_url)
+            .env("ALL_PROXY", &proxy_url)
+            .env("NO_PROXY", "")
+            .output()
+            .expect("run isolated object lambda proxy test child");
+
+        assert!(
+            output.status.success(),
+            "proxy test child failed: stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        target.join().expect("object lambda target should finish");
+    }
+
+    #[tokio::test]
+    async fn object_lambda_client_uses_configured_outbound_policy_resolver() {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind object lambda policy target");
+        let address = listener.local_addr().expect("object lambda policy target address");
+        let endpoint = Url::parse(&format!("http://{address}/transform")).expect("object lambda endpoint");
+        let policy = OutboundPolicy::from_allowed_origins(&endpoint.origin().ascii_serialization())
+            .expect("loopback test origin should be explicitly allowed");
+        let config = object_lambda_test_config_with_policy(endpoint.clone(), &policy);
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = listener.accept().await.expect("accept object lambda policy request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).await.expect("read object lambda policy request");
+            assert!(String::from_utf8_lossy(&request[..read]).starts_with("GET /transform HTTP/1.1"));
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .await
+                .expect("write object lambda policy response");
+        });
+
+        let response = build_object_lambda_http_client(&config)
+            .expect("object lambda policy client should build")
+            .get(endpoint)
+            .send()
+            .await
+            .expect("configured policy resolver should reach the allowed origin");
+        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+        server.await.expect("object lambda policy target should finish");
+    }
+
+    #[tokio::test]
+    async fn object_lambda_client_preserves_hostname_for_tls_sni() {
+        use rustls::{
+            ServerConfig, ServerConnection, StreamOwned,
+            pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer},
+        };
+        use std::io::{Read, Write};
+        use std::sync::{Arc, Once};
+
+        static INSTALL_CRYPTO_PROVIDER: Once = Once::new();
+        INSTALL_CRYPTO_PROVIDER.call_once(|| {
+            let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+        });
+        let rcgen::CertifiedKey { cert, signing_key } =
+            rcgen::generate_simple_self_signed(vec!["object-lambda.test".to_string()]).expect("cert should generate");
+        let temp_dir = tempfile::tempdir().expect("create temp directory");
+        let ca_path = temp_dir.path().join("object-lambda-ca.pem");
+        std::fs::write(&ca_path, cert.pem()).expect("write object lambda CA");
+        let server_config = Arc::new(
+            ServerConfig::builder()
+                .with_no_client_auth()
+                .with_single_cert(
+                    vec![cert.der().clone()],
+                    PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der())),
+                )
+                .expect("server certificate should be valid"),
+        );
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind object lambda TLS server");
+        let address = listener.local_addr().expect("object lambda TLS address");
+        let server = std::thread::spawn(move || {
+            let (stream, _) = listener.accept().expect("accept object lambda TLS request");
+            let connection = ServerConnection::new(server_config).expect("server TLS connection");
+            let mut stream = StreamOwned::new(connection, stream);
+            let mut request = [0_u8; 1024];
+            let _ = stream.read(&mut request).expect("read object lambda TLS request");
+            assert_eq!(stream.conn.server_name(), Some("object-lambda.test"));
+            stream
+                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("write object lambda TLS response");
+        });
+        let endpoint =
+            Url::parse(&format!("https://object-lambda.test:{}/transform", address.port())).expect("object lambda TLS endpoint");
+        let mut config = object_lambda_test_config(endpoint.clone());
+        config.client_ca = ca_path.to_string_lossy().into_owned();
+
+        let response = build_object_lambda_http_client_with_resolver(&config, StaticResolver(address.ip()))
+            .expect("object lambda TLS client should build")
+            .get(endpoint)
+            .send()
+            .await
+            .expect("TLS request should keep the configured hostname for SNI");
+        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
+        server.join().expect("object lambda TLS server should finish");
+    }
+
+    #[test]
+    fn object_lambda_tls_warning_redacts_endpoint_details() {
+        let captured = CapturedLog::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(captured.clone())
+            .finish();
+        let mut config = object_lambda_test_config(
+            Url::parse("https://object-lambda.test/private?token=secret").expect("object lambda endpoint"),
+        );
+        config.skip_tls_verify = true;
+
+        tracing::subscriber::with_default(subscriber, || {
+            build_object_lambda_http_client_with_resolver(&config, StaticResolver(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)))
+                .expect("object lambda client should build");
+        });
+
+        let logs = captured.contents();
+        assert!(logs.contains("https://object-lambda.test"));
+        for secret in ["/private", "token=secret"] {
+            assert!(!logs.contains(secret), "TLS warning leaked {secret}: {logs}");
+        }
+    }
+
+    #[tokio::test]
+    async fn object_lambda_request_error_does_not_expose_endpoint_details() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("reserve refused endpoint");
+        let address = listener.local_addr().expect("refused endpoint address");
+        drop(listener);
+        let request = reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("test client should build")
+            .get(format!("http://{address}/private?token=secret"));
+
+        let error = send_object_lambda_request(request)
+            .await
+            .expect_err("refused object lambda request should fail");
+        let rendered = error.to_string();
+        assert!(rendered.contains("object lambda target request failed"));
+        assert!(!rendered.contains("/private"));
+        assert!(!rendered.contains("token=secret"));
+    }
+
    #[test]
    fn build_object_lambda_get_request_removes_lambda_arn_and_preserves_request_inputs() {
        let mut req = S3Request {
@@ -323,9 +323,7 @@ pub(crate) mod replication {
    pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;
    pub(crate) type ReplicationStatusType = super::ecstore_bucket::replication::ReplicationStatusType;
    pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts;
-    #[cfg(test)]
    pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType;
-    #[cfg(test)]
    pub(crate) type TargetReplicationResyncStatus = super::ecstore_bucket::replication::TargetReplicationResyncStatus;

    pub(crate) fn resync_opts(
@@ -2089,3 +2089,78 @@ async fn put_object_computes_replication_decision_exactly_once() {
         pre-commit + post-commit recompute makes this observe 2 (rustfs/backlog#1320)"
    );
 }
+
+/// The object-lock handlers do not go through the object PUT path, so they must schedule
+/// replication themselves. Without it a retention or legal hold applied after upload stays
+/// local and the replica keeps its previous, unprotected state (a WORM object that is still
+/// deletable on the peer). This observes the shared decision counter: each handler must
+/// compute a replication decision exactly once, which is the step that drives both the
+/// persisted pending marker and the schedule.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
+async fn object_lock_handlers_schedule_replication() {
+    use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
+    use s3s::S3;
+    use std::sync::atomic::Ordering;
+
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let fs = FS::new();
+
+    let bucket = format!("test-lock-repl-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "locked/object.txt";
+
+    (*ecstore)
+        .make_bucket(
+            bucket.as_str(),
+            &MakeBucketOptions {
+                lock_enabled: true,
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("Failed to create object-lock bucket");
+    upload_test_object(&ecstore, bucket.as_str(), object, b"worm payload").await;
+
+    // PutObjectRetention must compute a replication decision.
+    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
+    let retention_input = PutObjectRetentionInput::builder()
+        .bucket(bucket.clone())
+        .key(object.to_string())
+        .retention(Some(ObjectLockRetention {
+            mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
+            retain_until_date: Some(time::macros::datetime!(2030-01-01 00:00:00 UTC).into()),
+        }))
+        .build()
+        .unwrap();
+    fs.put_object_retention(build_request(retention_input, Method::PUT))
+        .await
+        .expect("put_object_retention should succeed");
+    assert_eq!(
+        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
+        1,
+        "PutObjectRetention must compute the replication decision exactly once; 0 means the \
+         retention change never replicates and the peer copy stays unprotected"
+    );
+
+    // PutObjectLegalHold must do the same.
+    MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
+    let legal_hold_input = PutObjectLegalHoldInput::builder()
+        .bucket(bucket.clone())
+        .key(object.to_string())
+        .legal_hold(Some(ObjectLockLegalHold {
+            status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)),
+        }))
+        .build()
+        .unwrap();
+    fs.put_object_legal_hold(build_request(legal_hold_input, Method::PUT))
+        .await
+        .expect("put_object_legal_hold should succeed");
+    assert_eq!(
+        MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
+        1,
+        "PutObjectLegalHold must compute the replication decision exactly once; 0 means the \
+         legal hold never replicates and the peer copy stays deletable"
+    );
+}
@@ -262,6 +262,27 @@ fn secondary_tag_hint_action(action: Action, version_id: Option<&str>) -> Option
    }
 }

+/// GHSA-3ppv: select the IAM action for an object read by whether the request
+/// names an explicit object version. A read that targets a specific version must
+/// authorize against `s3:GetObjectVersion`; a current-object read authorizes
+/// against `s3:GetObject`. Reading a historical version while only holding
+/// `s3:GetObject` is an information-disclosure bypass, so the two must not be
+/// conflated at the authorization boundary.
+///
+/// Note: `ActionSet::is_match` still maps a `s3:GetObjectVersion` grant onto a
+/// `s3:GetObject` request. That mapping is intentionally left in place until the
+/// remaining version-aware read paths (HeadObject, GetObjectAcl, tagging) get the
+/// same treatment — see the GHSA-3ppv follow-up audit. It does not re-open this
+/// disclosure: it only broadens a Version grant toward current reads, never the
+/// reverse.
+fn versioned_read_action(version_id: Option<&str>) -> Action {
+    if version_id.is_some() {
+        Action::S3Action(S3Action::GetObjectVersionAction)
+    } else {
+        Action::S3Action(S3Action::GetObjectAction)
+    }
+}
+
 async fn get_or_fetch_object_tag_conditions<T>(
    req: &mut S3Request<T>,
    bucket: &str,
@@ -1025,7 +1046,9 @@ impl S3Access for FS {
            req_info.object = Some(src_key.clone());
            req_info.version_id = version_id.clone();

-            authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
+            // GHSA-3ppv: a versioned copy source must authorize against
+            // s3:GetObjectVersion, not s3:GetObject.
+            authorize_request(req, versioned_read_action(version_id.as_deref())).await?;
        }

        let req_info = ext_req_info_mut(&mut req.extensions)?;
@@ -1470,7 +1493,9 @@ impl S3Access for FS {
        req_info.object = Some(req.input.key.clone());
        req_info.version_id = req.input.version_id.clone();

-        authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await
+        // GHSA-3ppv: a versioned read (?versionId=...) must authorize against
+        // s3:GetObjectVersion, not s3:GetObject.
+        authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await
    }

    /// Checks whether the GetObjectAcl request has accesses to the resources.
@@ -2050,7 +2075,9 @@ impl S3Access for FS {
            req_info.object = Some(src_key.clone());
            req_info.version_id = version_id.clone();

-            authorize_request(req, Action::S3Action(S3Action::GetObjectAction)).await?;
+            // GHSA-3ppv: a versioned copy source must authorize against
+            // s3:GetObjectVersion, not s3:GetObject.
+            authorize_request(req, versioned_read_action(version_id.as_deref())).await?;
        }

        let req_info = ext_req_info_mut(&mut req.extensions)?;
@@ -2080,7 +2107,7 @@ mod tests {
        legal_hold_write_requested, list_parts_authorize_action, load_bucket_policy_existing_object_tag_hint,
        merge_list_bucket_query_conditions, owner_can_bypass_policy_deny, post_object_authorize_action,
        put_bucket_policy_authorize_action, request_context_from_req, retention_write_requested, secondary_tag_hint_action,
-        table_data_plane_admin_action, validate_post_object_success_controls,
+        table_data_plane_admin_action, validate_post_object_success_controls, versioned_read_action,
    };
    use crate::error::ApiError;
    use http::{Extensions, HeaderMap, Method, Uri};
@@ -2113,6 +2140,29 @@ mod tests {
        assert_eq!(get_bucket_policy_authorize_action(), Action::S3Action(S3Action::GetBucketPolicyAction));
    }

+    #[test]
+    fn ghsa_3ppv_versioned_read_selects_get_object_version_action() {
+        // Regression (GHSA-3ppv): a read that names an explicit version must
+        // authorize against s3:GetObjectVersion so that a principal holding only
+        // s3:GetObject cannot read historical versions. Applies to GetObject and
+        // the CopyObject / UploadPartCopy sources.
+        assert_eq!(
+            versioned_read_action(Some("0194e0f1-0000-7000-8000-000000000000")),
+            Action::S3Action(S3Action::GetObjectVersionAction),
+            "an explicit versionId must require GetObjectVersion"
+        );
+        assert_eq!(
+            versioned_read_action(Some("null")),
+            Action::S3Action(S3Action::GetObjectVersionAction),
+            "the sentinel `null` version is still an explicit version selector"
+        );
+        assert_eq!(
+            versioned_read_action(None),
+            Action::S3Action(S3Action::GetObjectAction),
+            "a current-object read (no versionId) authorizes against GetObject"
+        );
+    }
+
    #[test]
    fn put_bucket_policy_uses_put_bucket_policy_action() {
        assert_eq!(put_bucket_policy_authorize_action(), Action::S3Action(S3Action::PutBucketPolicyAction));
@@ -45,6 +45,7 @@ use rustfs_targets::EventName;
 use rustfs_utils::http::headers::{
    AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
 };
+use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str};
 use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error};
 use std::fmt::Debug;
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
@@ -56,6 +57,9 @@ const LOG_SUBSYSTEM_OBJECT: &str = "object";
 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_object, schedule_object_replication,
+};
 use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;

 #[derive(Debug, Clone)]
@@ -1282,20 +1286,53 @@ impl S3 for FS {
            .await
            .map_err(ApiError::from)?;

-        let eval_metadata = parse_object_lock_legal_hold(legal_hold)?;
-
-        let popts = ObjectOptions {
+        let mut popts = ObjectOptions {
            mod_time: opts.mod_time,
-            version_id: opts.version_id,
-            eval_metadata: Some(eval_metadata),
+            version_id: opts.version_id.clone(),
            ..Default::default()
        };

+        // PutObjectLegalHold only rewrites metadata, so replication is not scheduled by the
+        // object PUT path. Schedule it explicitly, otherwise the legal hold never reaches the
+        // replica and the peer copy remains deletable.
+        //
+        // Mirror the PUT path ordering (see object_usecase::put_object): compute the decision
+        // once BEFORE the commit and persist the pending marker with it, so a restart between
+        // the commit and the worker write-back leaves a Pending status on disk for the scanner
+        // to re-drive. Scheduling without that marker would lose the task silently.
+        let dsc = match store.get_object_info(&bucket, &key, &opts).await.ok() {
+            Some(info) => {
+                must_replicate_object(
+                    &bucket,
+                    &key,
+                    &info.user_defined,
+                    info.user_tags.as_ref().clone(),
+                    popts.delete_marker_replication_status(),
+                    popts.clone(),
+                )
+                .await
+            }
+            None => ReplicateDecision::new(),
+        };
+
+        let mut eval_metadata = parse_object_lock_legal_hold(legal_hold)?;
+        if dsc.replicate_any() {
+            insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+            insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
+        }
+        popts.eval_metadata = Some(eval_metadata);
+
        let info = store.put_object_metadata(&bucket, &key, &popts).await.map_err(|e| {
            error!("put_object_metadata failed, {}", e.to_string());
            s3_error!(InternalError, "{}", e.to_string())
        })?;

+        // This replicates the whole object, not just the changed metadata: metadata-only
+        // replication is not implemented yet, so the lock change triggers a full re-upload.
+        if dsc.replicate_any() {
+            schedule_object_replication(info.clone(), store.clone(), dsc).await;
+        }
+
        let output = PutObjectLegalHoldOutput {
            request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)),
        };
@@ -1438,7 +1475,9 @@ impl S3 for FS {
            .await
            .map_err(ApiError::from)?;

-        if let Ok(existing_obj_info) = store.get_object_info(&bucket, &key, &check_opts).await
+        let existing_obj_info = store.get_object_info(&bucket, &key, &check_opts).await.ok();
+
+        if let Some(existing_obj_info) = existing_obj_info.as_ref()
            && let Some(block_reason) = check_retention_for_modification(
                &existing_obj_info.user_defined,
                new_mode.as_deref(),
@@ -1449,23 +1488,57 @@ impl S3 for FS {
            return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message()));
        }

-        let eval_metadata = parse_object_lock_retention(retention)?;
-
        let mut opts: ObjectOptions = get_opts(&bucket, &key, version_id, None, &req.headers)
            .await
            .map_err(ApiError::from)?;
-        opts.eval_metadata = Some(eval_metadata);
        opts.object_lock_retention = Some(ObjectLockRetentionOptions {
            mode: new_mode,
            retain_until: new_retain_until,
            bypass_governance,
        });

+        // PutObjectRetention only rewrites metadata, so the object PUT path that normally
+        // computes a replication decision and schedules replication never runs for it. Without
+        // scheduling here the peer keeps the previous, unprotected lock state and a
+        // WORM-protected object stays deletable on the replica.
+        //
+        // Mirror the PUT path ordering (see object_usecase::put_object): compute the decision
+        // once BEFORE the commit and persist the pending marker with it, so a restart between
+        // the commit and the worker write-back leaves a Pending status on disk for the scanner
+        // to re-drive. Scheduling without that marker would lose the task silently.
+        let dsc = match existing_obj_info.as_ref() {
+            Some(info) => {
+                must_replicate_object(
+                    &bucket,
+                    &key,
+                    &info.user_defined,
+                    info.user_tags.as_ref().clone(),
+                    opts.delete_marker_replication_status(),
+                    opts.clone(),
+                )
+                .await
+            }
+            None => ReplicateDecision::new(),
+        };
+
+        let mut eval_metadata = parse_object_lock_retention(retention)?;
+        if dsc.replicate_any() {
+            insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+            insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default());
+        }
+        opts.eval_metadata = Some(eval_metadata);
+
        let object_info = store.put_object_metadata(&bucket, &key, &opts).await.map_err(|e| {
            error!("put_object_metadata failed, {}", e.to_string());
            S3Error::from(ApiError::from(e))
        })?;

+        // This replicates the whole object, not just the changed metadata: metadata-only
+        // replication is not implemented yet, so the lock change triggers a full re-upload.
+        if dsc.replicate_any() {
+            schedule_object_replication(object_info.clone(), store.clone(), dsc).await;
+        }
+
        let output = PutObjectRetentionOutput {
            request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)),
        };
Author	SHA1	Message	Date
Zhengchao An	1c8088d0b2	fix(security): redact OIDC secrets from logs and errors (#5147 ) Stop the OIDC subsystem from writing credential-grade secrets into logs and returned errors. - crates/iam/src/oidc.rs: redact sensitive header values (authorization, proxy-authorization, cookie, set-cookie) in format_http_headers, emitting only name and length; drop the raw request/response body from the DEBUG events (keep byte length); stop logging and stop splicing the raw token response body into the error returned on token_response_parse_failed (the TokenResponseBodyShape summary and length are retained); remove the now-unused format_http_body helper. - rustfs/src/admin/handlers/oidc.rs: stop logging the raw authorization code and state on the code-exchange error path (code_len/state_len are kept). This is the OIDC log/error redaction pre-work (batch 0B) from the OIDC review in rustfs/backlog#1437. It changes diagnostic content only; HTTP/STS status codes and legitimate request results are unchanged.	2026-07-23 01:09:53 +00:00
Zhengchao An	ffcdab900a	chore(release): prepare 1.0.0-beta.11 (#5146 ) Bump workspace version and release assets to 1.0.0-beta.11.	2026-07-23 01:09:29 +00:00
Zhengchao An	b94bf874bf	test(admin): pin GHSA-5354 scope guard for derived credentials (#5144 ) * fix(admin): confine service-account parent to caller scope (GHSA-5354) AddServiceAccount gated creation only on CreateServiceAccountAdminAction and did not constrain the targetUser (the new service account's parent) to the caller's own scope. Combined with the deliberate root-credential exemption in the existence check, a non-owner principal holding that admin action could create a service account parented to the root credential, which then authenticates as owner via prepare_service_account_auth. This is the direct-handler analogue of the ImportIam parent-scope check (GHSA-566f), which already enforces the invariant via imported_service_account_parent_allowed. Enforce the same invariant through add_service_account_parent_within_scope: a non-owner may create a service account only for itself (or, for a derived credential, its own parent); owners retain cross-user creation. Add named regression test ghsa_5354_non_owner_service_account_parent_confined_to_scope. Refs GHSA-5354-r3w2-34m8. * test(admin): pin GHSA-5354 scope guard for derived credentials The GHSA-5354 fix (#5141) added `add_service_account_parent_within_scope` and a named-boundary test, but that test only covers the self-scoped case with `req_user == req_parent_user`. A derived credential — a service account or STS token holding CreateServiceAccountAdminAction — has `req_user` (its own key) distinct from `req_parent_user` (its parent), which is the realistic attacker shape and was left unexercised. Add a test that pins the guard's allow set to exactly `owner \|\| is_svc_acc` across that derived-credential shape, including a derived non-owner aiming at the root credential, so a later change to either the guard or the `is_svc_acc` rewrite cannot silently let a derived non-owner escape its own parent's scope. Also note at the check site that the guard is evaluated on the original `target_user`, before the derived-credential rewrite to `req_parent_user`. Refs GHSA-5354-r3w2-34m8.	2026-07-23 08:48:28 +08:00
Zhengchao An	7d96ffd7fb	docs(testing): record GHSA-5354 and GHSA-3ppv regression guards (#5143 ) PRs #5141 and #5142 fixed GHSA-5354 (service-account parent-scope enforcement) and GHSA-3ppv (versioned object-read authorization) but did not add the advisories to the security-regressions inventory that repo policy requires for every fixed GHSA. Add both rows to the advisory -> test mapping (advisory, class, fix PR, named regression test, layer) and list ghsa_5354_* / ghsa_3ppv_* under the default-CI unit-test execution map, since both tests live in the rustfs lib target and run in the default `cargo nextest run --profile ci --all --exclude e2e_test` pass.	2026-07-23 08:31:21 +08:00
cxymds	027a749646	feat(admin): advertise site replication capabilities (#5131 )	2026-07-23 08:02:50 +08:00
cxymds	6c23b8506e	feat(site-replication): persist durable resync lifecycle (#5125 )	2026-07-23 08:02:30 +08:00
Zhengchao An	8166561702	fix(s3): authorize versioned object reads against GetObjectVersion (GHSA-3ppv) (#5142 ) get_object, the CopyObject source, and the UploadPartCopy source authorized reads that name an explicit versionId against s3:GetObject rather than s3:GetObjectVersion. A principal holding s3:GetObject but not s3:GetObjectVersion could therefore read historical object versions. get_object_attributes was already version-aware; these three paths were not. Add a shared versioned_read_action helper that selects s3:GetObjectVersion when a version is named and s3:GetObject otherwise, and apply it to the three read/copy-source authorization sites. Add named regression test ghsa_3ppv_versioned_read_selects_get_object_version_action. The ActionSet::is_match GetObjectVersion->GetObject compatibility mapping is intentionally retained for now; its removal is gated on a follow-up audit of the remaining version-aware read paths (HeadObject, GetObjectAcl, tagging). It does not affect this fix, which only prevents a GetObject grant from satisfying a versioned read. Refs GHSA-3ppv-fx5m-m749.	2026-07-22 23:44:05 +00:00
Zhengchao An	9866f68d86	fix(admin): confine service-account parent to caller scope (GHSA-5354) (#5141 ) AddServiceAccount gated creation only on CreateServiceAccountAdminAction and did not constrain the targetUser (the new service account's parent) to the caller's own scope. Combined with the deliberate root-credential exemption in the existence check, a non-owner principal holding that admin action could create a service account parented to the root credential, which then authenticates as owner via prepare_service_account_auth. This is the direct-handler analogue of the ImportIam parent-scope check (GHSA-566f), which already enforces the invariant via imported_service_account_parent_allowed. Enforce the same invariant through add_service_account_parent_within_scope: a non-owner may create a service account only for itself (or, for a derived credential, its own parent); owners retain cross-user creation. Add named regression test ghsa_5354_non_owner_service_account_parent_confined_to_scope. Refs GHSA-5354-r3w2-34m8.	2026-07-22 23:43:30 +00:00
abdullahnah92	0e8f1a187c	fix(site-replication): replicate object-lock retention and legal hold (#5105 ) * fix(site-replication): replicate object-lock retention and legal hold Object Lock did not survive site replication, so a WORM-protected object was either missing from the peer entirely or present there unprotected. Two independent defects caused this. 1. The site-replicator service account policy granted no object-lock actions. Any replicated object carrying x-amz-object-lock-mode / x-amz-object-lock-retain-until-date was rejected by the peer with AccessDenied, so an object created under a retention rule never replicated at all. Grant s3:GetObjectRetention, s3:PutObjectRetention, s3:GetObjectLegalHold and s3:PutObjectLegalHold. s3:BypassGovernanceRetention is deliberately NOT granted: replication must never be able to erase a retained version on the peer. 2. PutObjectRetention and PutObjectLegalHold only rewrote object metadata. The object PUT path is what normally computes a replication decision and schedules replication, so a lock applied after upload stayed local and the peer kept its previous, unprotected state. Schedule replication explicitly after both metadata writes. Verified live on a two-site cluster: an object created under a COMPLIANCE retention rule now replicates, the peer copy reports COMPLIANCE, a version-scoped delete on the peer is refused, and the content matches by MD5. Retention and legal hold applied after upload propagate to the peer within about 15s, and the replicated copy is likewise undeletable. No AccessDenied replication errors remain. Adds a regression test asserting the policy allows the four object-lock actions and still denies governance bypass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(site-replication): persist pending marker before object-lock metadata commit Review follow-up on the object-lock replication fix. PutObjectRetention and PutObjectLegalHold scheduled replication but did not persist a pending replication marker first, unlike the canonical object PUT path. If the process restarted between the metadata commit and the replication worker write-back, the in-memory task was lost and the object carried no Pending/Failed marker, so the scanner heal skipped it and the peer silently stayed unprotected. Both handlers now mirror the PUT path ordering: compute the replication decision once BEFORE the commit, fold SUFFIX_REPLICATION_STATUS and SUFFIX_REPLICATION_TIMESTAMP into the eval_metadata passed to put_object_metadata, then schedule after the commit. eval_metadata is merged into the object metadata, so the lock fields are preserved. When the object info cannot be read the decision is empty and nothing is scheduled. Also aligns the two handlers to build their options the same way, and notes at both schedule sites that this triggers a full object re-upload because metadata-only replication is not implemented yet. Adds an integration test asserting both handlers compute a replication decision exactly once, so a regression that drops the scheduling is caught. Verified live on a two-site cluster: applying retention while the peer is down now leaves Replication Status: PENDING on disk at commit time, and once the peer returns the lock propagates in about 15s and the status becomes COMPLETED. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-23 07:34:22 +08:00
Zhengchao An	b32bd1f8a9	fix(ecstore): cap mmap-copy shard reads to stop large single-part GET OOM (#5140 )	2026-07-22 23:32:29 +00:00
Zhengchao An	c9848a6096	fix(security): enforce outbound connection policy (#5135 ) Co-authored-by: cxymds <cxymds@gmail.com>	2026-07-22 22:11:28 +00:00
Zhengchao An	6e88ab2a25	fix(targets): accept both DnsFailure and Unreachable in macOS DNS test (#5138 ) On macOS with DNS interception services, .invalid TLD domains resolve to an interception address (e.g. 198.18.16.173) instead of failing DNS resolution. The health probe then classifies the error as Unreachable rather than DnsFailure. Accept both outcomes since both are correct non-reachable error classifications.	2026-07-23 03:02:47 +08:00
houseme	05d4480f08	fix(admin): preserve peer topology slots (#5136 ) Keep remote peer topology slots observable when peer client construction cannot build a dialing client, and publish admin server_info cache/failure state only after a complete probe round. Covers rustfs/backlog#1426 and rustfs/backlog#1430. Co-authored-by: heihutu <heihutu@gmail.com>	2026-07-22 18:02:41 +00:00
houseme	abee09dad9	fix(admin): preserve v3 topology membership (#5134 ) * fix(admin): use raw topology host for peer mapping Backlog: rustfs/backlog#1427 Keep endpoint topology membership keyed by raw host:port when mapping peers to grid hosts, while preserving DNS resolution only for the PeerRestClient dialing host. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(admin): reconcile v3 info with topology Backlog: rustfs/backlog#1425 Backlog: rustfs/backlog#1428 Synthesize additive unknown server rows from configured endpoint topology before v3 backend counters are computed, and add exact drive-identity coverage tests so balanced totals cannot hide a missing member. Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(admin): collapse topology reconciliation Build a single endpoint topology index for v3 admin server reconciliation and completeness reporting, and reuse the raw peer/grid mapping when constructing peer clients. Backlog: rustfs/backlog#1424 Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>	2026-07-22 15:31:31 +00:00