volume: replicate FetchAndWriteNeedle writes over the configured TLS scheme (#12128)

* volume: replicate FetchAndWriteNeedle writes with the configured TLS scheme

The guarded upload path hardcoded an http:// URL and used a bare
*http.Client, so a peer fronted by https rejected the replica write and
read-through of a remote-evicted object failed whenever the volume had
replicas. Normalize the upload URL through the shared global client and
let the guarded transport borrow its https.client TLS configuration, the
same settings the regular replication path resolves.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: keep the global HTTPS downgrade guard on replica uploads

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: cover TLS replicate uploads and the redirect downgrade guard

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: pin the downgrade guard against a dialable plaintext target

The previous redirect pointed at 169.254.169.254, which the guarded dialer
already refuses, so the test passed even with CheckRedirect removed.
Redirect to a live plaintext server on a dialable peer address and assert
it never receives the resent body.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume-rs: keep the HTTPS downgrade redirect guard on outbound writes

Mirrors the Go peer upload client: the shared outgoing HTTP client now stops redirects that would downgrade an https request to plaintext, so replicate uploads to TLS peers cannot leak data or auth headers via a 307 to http.

* volume-rs: fail rejected redirects instead of returning the 30x response

Go's rejectHttpsDowngrade uses ErrUseLastResponse, which surfaces to the
caller as a non-2xx status it rejects; attempt.stop() returned the redirect
response as Ok, so a refused downgrade could masquerade as a completed write
to any caller that does not check the status.

* volume-rs: fail replica writes on a non-2xx status

fetch_and_write_needle discarded the replica response without checking its
status, so a peer rejection — or a redirect the client refused to follow —
was reported back as a successful copy while no replica was written. Go's
doUploadData fails on any non-success status.

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/main.rs

@@ -102,6 +102,22 @@ fn main() {
     }
 }
 
+/// Mirrors Go's rejectHttpsDowngrade: a redirect must not downgrade a request
+/// carrying data and auth headers from https to plaintext.
+fn redirect_is_https_downgrade(previous: &[reqwest::Url], next: &reqwest::Url) -> bool {
+    matches!(previous.first(), Some(url) if url.scheme() == "https") && next.scheme() != "https"
+}
+
+fn reject_https_downgrade(attempt: reqwest::redirect::Attempt) -> reqwest::redirect::Action {
+    if attempt.previous().len() >= 10 {
+        attempt.error("redirect limit reached")
+    } else if redirect_is_https_downgrade(attempt.previous(), attempt.url()) {
+        attempt.error("redirect downgraded https to plaintext")
+    } else {
+        attempt.follow()
+    }
+}
+
 fn build_outgoing_http_client(
     config: &VolumeServerConfig,
 ) -> Result<(reqwest::Client, String), Box<dyn std::error::Error>> {
@@ -110,11 +126,17 @@ fn build_outgoing_http_client(
     } else {
         "http"
     };
+    let redirect_policy = reqwest::redirect::Policy::custom(reject_https_downgrade);
     if !config.https_client_enabled {
-        return Ok((reqwest::Client::new(), scheme.to_string()));
+        return Ok((
+            reqwest::Client::builder()
+                .redirect(redirect_policy)
+                .build()?,
+            scheme.to_string(),
+        ));
     }
 
-    let mut builder = reqwest::Client::builder();
+    let mut builder = reqwest::Client::builder().redirect(redirect_policy);
     if !config.https_client_ca_file.is_empty() {
         let ca_pem = std::fs::read(&config.https_client_ca_file).map_err(|e| {
             format!(
@@ -1152,4 +1174,19 @@ mod tests {
             cipher_suites: String::new(),
         }));
     }
+
+    #[test]
+    fn test_redirect_is_https_downgrade() {
+        use super::redirect_is_https_downgrade;
+        let https = reqwest::Url::parse("https://peer:8443/1?type=replicate").unwrap();
+        let http = reqwest::Url::parse("http://other:8080/1?type=replicate").unwrap();
+        assert!(redirect_is_https_downgrade(&[https.clone()], &http));
+        assert!(!redirect_is_https_downgrade(&[http.clone()], &https));
+        assert!(!redirect_is_https_downgrade(&[http.clone()], &http));
+        assert!(!redirect_is_https_downgrade(&[https.clone()], &https));
+        assert!(!redirect_is_https_downgrade(&[], &http));
+        // Only the origin scheme counts (Go's via[0]): a chain that started
+        // plaintext loses nothing by returning to plaintext.
+        assert!(!redirect_is_https_downgrade(&[http.clone(), https.clone()], &http));
+    }
 }

seaweed-volume/src/server/grpc_server.rs

@@ -5076,15 +5076,25 @@ impl VolumeServer for VolumeGrpcService {
                 handles.push(tokio::spawn(async move {
                     let form = reqwest::multipart::Form::new()
                         .part("file", reqwest::multipart::Part::bytes(data_clone));
-                    client_clone
+                    let resp = client_clone
                         .post(&url)
                         .multipart(form)
                         .send()
                         .await
-                        .map(|_| ())
                         .map_err(|e| {
                             format!("remote write needle {} size {}: {}", needle_id, size, e)
-                        })
+                        })?;
+                    if resp.status().is_success() {
+                        Ok(())
+                    } else {
+                        Err(format!(
+                            "remote write needle {} size {}: {} returned status {}",
+                            needle_id,
+                            size,
+                            url,
+                            resp.status()
+                        ))
+                    }
                 }));
             }
         }

weed/server/volume_grpc_remote.go

@@ -28,6 +28,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
 // lookupIPAddrFunc resolves a host to one or more IP addresses. It is a
@@ -311,6 +312,20 @@ func newGuardedHTTPClientPolicy(endpoint string, allowPrivate bool) *http.Client
 	}
 }
 
+// newGuardedPeerHTTPClient is newGuardedHTTPClientPolicy for uploads to cluster
+// peers: a volume server fronted by TLS expects the https.client configuration
+// the shared global client carries, so the guarded transport borrows it along
+// with its redirect policy — a redirect must not downgrade an upload carrying
+// data and auth headers to plaintext.
+func newGuardedPeerHTTPClient(endpoint string) *http.Client {
+	globalClient := util_http.GetGlobalHttpClient()
+	client := newGuardedHTTPClientPolicy(endpoint, true)
+	client.Transport.(*http.Transport).TLSClientConfig =
+		globalClient.GetClientTransport().TLSClientConfig
+	client.CheckRedirect = globalClient.Client.CheckRedirect
+	return client
+}
+
 // guardedRemoteClient reports the caller-supplied endpoint a backend dials
 // directly and a constructor that routes through the given HTTP client, or
 // ok=false when nothing in the conf steers a destination. The S3-SDK family,
@@ -573,8 +588,14 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 			wg.Add(1)
 			go func(idx int, targetVolumeServer string) {
 				defer wg.Done()
+				uploadUrl, urlErr := util_http.GetGlobalHttpClient().NormalizeHttpScheme(
+					fmt.Sprintf("%s/%s?type=replicate", targetVolumeServer, fileId.String()))
+				if urlErr != nil {
+					replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, urlErr)
+					return
+				}
 				uploadOption := &operation.UploadOption{
-					UploadUrl:         fmt.Sprintf("http://%s/%s?type=replicate", targetVolumeServer, fileId.String()),
+					UploadUrl:         uploadUrl,
 					Filename:          "",
 					Cipher:            false,
 					IsInputCompressed: false,
@@ -597,7 +618,7 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 						return
 					}
 				} else {
-					uploader = operation.NewUploaderWithHttpClient(newGuardedHTTPClientPolicy(targetVolumeServer, true))
+					uploader = operation.NewUploaderWithHttpClient(newGuardedPeerHTTPClient(targetVolumeServer))
 				}
 
 				if _, replicaWriteErr := uploader.UploadData(ctx, data, uploadOption); replicaWriteErr != nil {

weed/server/volume_grpc_remote_test.go

@@ -2,16 +2,29 @@ package weed_server
 
 import (
 	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
 	"errors"
+	"io"
+	"math/big"
 	"net"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"strings"
 	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	s3remote "github.com/seaweedfs/seaweedfs/weed/remote_storage/s3"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
 // stubLookup returns a resolver func that maps the supplied hostnames to
@@ -779,3 +792,181 @@ func TestValidateRemoteConfForLoad(t *testing.T) {
 		t.Error("aliyun with empty endpoint should be rejected by the guard")
 	}
 }
+
+// TestGuardedPeerHTTPClientCarriesClusterTLS confirms the replica-upload
+// client borrows the shared global client's TLS configuration — an https
+// volume server expects the https.client cert and CA — while the remote
+// storage client stays on a plain transport.
+func TestGuardedPeerHTTPClientCarriesClusterTLS(t *testing.T) {
+	globalTransport := util_http.GetGlobalHttpClient().GetClientTransport()
+	original := globalTransport.TLSClientConfig
+	globalTransport.TLSClientConfig = &tls.Config{ServerName: "volume-peer.test"}
+	t.Cleanup(func() { globalTransport.TLSClientConfig = original })
+
+	peer := newGuardedPeerHTTPClient("10.0.0.1:8080")
+	if peer.Transport.(*http.Transport).TLSClientConfig != globalTransport.TLSClientConfig {
+		t.Error("peer upload client should share the global TLS config")
+	}
+	if peer.CheckRedirect == nil {
+		t.Error("peer upload client should share the global redirect policy")
+	}
+	if remote := newGuardedHTTPClient("10.0.0.1:8080").Transport.(*http.Transport); remote.TLSClientConfig != nil {
+		t.Error("remote storage client should not carry cluster TLS config")
+	}
+}
+
+// peerIPv4 returns a non-loopback IPv4 of this host — the only kind of
+// address the peer dialer connects to — or skips the test when none exists.
+func peerIPv4(t *testing.T) net.IP {
+	t.Helper()
+
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		t.Skipf("list interface addresses: %v", err)
+	}
+	for _, addr := range addrs {
+		ipnet, ok := addr.(*net.IPNet)
+		if !ok {
+			continue
+		}
+		if ip := ipnet.IP.To4(); ip != nil && !ip.IsLoopback() && !ip.IsLinkLocalUnicast() {
+			return ip
+		}
+	}
+	t.Skip("no non-loopback IPv4 for the peer dialer to reach")
+	return nil
+}
+
+// peerTLSServer runs handler over TLS on a non-loopback IPv4 under a
+// self-signed cert carrying that IP, and returns its host:port plus a pool
+// trusting the cert.
+func peerTLSServer(t *testing.T, handler http.Handler) (string, *x509.CertPool) {
+	t.Helper()
+
+	peerIP := peerIPv4(t)
+
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+	template := &x509.Certificate{
+		SerialNumber:          big.NewInt(1),
+		Subject:               pkix.Name{CommonName: "volume-peer"},
+		NotBefore:             time.Now().Add(-time.Hour),
+		NotAfter:              time.Now().Add(time.Hour),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+		IPAddresses:           []net.IP{peerIP},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
+	if err != nil {
+		t.Fatalf("create certificate: %v", err)
+	}
+	parsed, err := x509.ParseCertificate(der)
+	if err != nil {
+		t.Fatalf("parse certificate: %v", err)
+	}
+	pool := x509.NewCertPool()
+	pool.AddCert(parsed)
+
+	listener, err := net.Listen("tcp", net.JoinHostPort(peerIP.String(), "0"))
+	if err != nil {
+		t.Skipf("bind non-loopback %s: %v", peerIP, err)
+	}
+	server := &httptest.Server{
+		Listener: listener,
+		Config:   &http.Server{Handler: handler},
+		TLS:      &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}},
+	}
+	server.StartTLS()
+	t.Cleanup(server.Close)
+	return listener.Addr().String(), pool
+}
+
+// usePeerCA points the shared global client's transport at pool for the test.
+func usePeerCA(t *testing.T, pool *x509.CertPool) {
+	t.Helper()
+	transport := util_http.GetGlobalHttpClient().GetClientTransport()
+	original := transport.TLSClientConfig
+	transport.TLSClientConfig = &tls.Config{RootCAs: pool}
+	t.Cleanup(func() { transport.TLSClientConfig = original })
+}
+
+// TestGuardedPeerHTTPClientUploadsOverTLS sends a real replicate upload
+// through the guarded peer client to a TLS server — the global client's CA
+// config and the https scheme must carry it, so a hardcoded http:// URL
+// could never reach the handler.
+func TestGuardedPeerHTTPClientUploadsOverTLS(t *testing.T) {
+	var sawReplicate atomic.Bool
+	addr, pool := peerTLSServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.TLS == nil {
+			t.Error("peer upload arrived over plaintext")
+		}
+		sawReplicate.Store(r.URL.Query().Get("type") == "replicate")
+		io.Copy(io.Discard, r.Body)
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	usePeerCA(t, pool)
+
+	uploader := operation.NewUploaderWithHttpClient(newGuardedPeerHTTPClient(addr))
+	data := []byte("needle-data")
+	result, err := uploader.UploadData(t.Context(), data, &operation.UploadOption{
+		UploadUrl:     "https://" + addr + "/1,01637037d6?type=replicate",
+		IsReplication: true,
+	})
+	if err != nil {
+		t.Fatalf("upload to TLS peer: %v", err)
+	}
+	if !sawReplicate.Load() {
+		t.Error("peer never saw a ?type=replicate upload")
+	}
+	if result == nil || result.Size != uint32(len(data)) {
+		t.Errorf("unexpected upload result %+v", result)
+	}
+}
+
+// TestGuardedPeerHTTPClientRejectsHTTPSDowngrade pins the downgrade guard the
+// global client carries: an HTTPS peer answering with a redirect to a
+// reachable http server must fail the upload rather than resend the body and
+// auth header in plaintext. The plaintext target sits on an address the peer
+// dialer accepts, so only the redirect policy can stop the resend.
+func TestGuardedPeerHTTPClientRejectsHTTPSDowngrade(t *testing.T) {
+	var sawPlaintext atomic.Bool
+	plaintext := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sawPlaintext.Store(true)
+		io.Copy(io.Discard, r.Body)
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	listener, err := net.Listen("tcp", net.JoinHostPort(peerIPv4(t).String(), "0"))
+	if err != nil {
+		t.Skipf("bind non-loopback plaintext target: %v", err)
+	}
+	plaintext.Listener = listener
+	plaintext.Start()
+	defer plaintext.Close()
+
+	var sawUpload atomic.Bool
+	addr, pool := peerTLSServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sawUpload.Store(true)
+		io.Copy(io.Discard, r.Body)
+		http.Redirect(w, r, "http://"+listener.Addr().String()+"/latest/meta-data", http.StatusTemporaryRedirect)
+	}))
+	usePeerCA(t, pool)
+
+	uploader := operation.NewUploaderWithHttpClient(newGuardedPeerHTTPClient(addr))
+	if _, err := uploader.UploadData(t.Context(), []byte("needle-data"), &operation.UploadOption{
+		UploadUrl:     "https://" + addr + "/1,01637037d6?type=replicate",
+		IsReplication: true,
+		MaxAttempts:   1,
+	}); err == nil {
+		t.Error("HTTPS→HTTP redirect must fail the upload, not resend the body in plaintext")
+	}
+	if sawPlaintext.Load() {
+		t.Error("redirect resend reached the plaintext server")
+	}
+	if !sawUpload.Load() {
+		t.Error("HTTPS peer never saw the original upload")
+	}
+}