From 7a47c63d540e798eb5139b81343824d608944c57 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 27 Oct 2026 09:09:04 -0700 Subject: [PATCH] volume: replicate FetchAndWriteNeedle writes over the configured TLS scheme (#12128) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 41 ++++- seaweed-volume/src/server/grpc_server.rs | 16 +- weed/server/volume_grpc_remote.go | 25 ++- weed/server/volume_grpc_remote_test.go | 191 +++++++++++++++++++++++ 4 files changed, 266 insertions(+), 7 deletions(-) diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 46fe45289d..ace64ddc07 100644 --- a/seaweed-volume/src/main.rs +++ b/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> { @@ -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)); + } } diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 88dddcbbf6..e0b4507914 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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() + )) + } })); } } diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index de82b65eee..adeb59e5de 100644 --- a/weed/server/volume_grpc_remote.go +++ b/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 { diff --git a/weed/server/volume_grpc_remote_test.go b/weed/server/volume_grpc_remote_test.go index 03bcf150d7..cdee1f8db1 100644 --- a/weed/server/volume_grpc_remote_test.go +++ b/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") + } +}