fix(sync): use per-cluster TLS for HTTP volume connections in filer.sync (#8974)

* fix(sync): use per-cluster TLS for HTTP volume connections in filer.sync (#8965)

When filer.sync runs with -a.security and -b.security flags, only gRPC
connections received per-cluster TLS configuration. HTTP clients for
volume server reads and uploads used a global singleton with the default
security.toml, causing TLS verification failures when clusters use
different self-signed certificates.

Load per-cluster HTTPS client config from the security files and pass
dedicated HTTP clients to FilerSource (for downloads) and FilerSink
(for uploads) so each direction uses the correct cluster's certificates.

* fix(sync): address review feedback for per-cluster HTTP TLS

- Add insecure_skip_verify support to NewHttpClientWithTLS and read it
  from per-cluster security config via https.client.insecure_skip_verify
- Error on partial mTLS config (cert without key or vice versa)
- Add nil-check for client parameter in DownloadFileWithClient
- Document SetUploader as init-only (same pattern as SetChunkConcurrency)

weed/command/filer_sync.go

@@ -15,12 +15,14 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink/filersink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	statsCollect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/grace"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"google.golang.org/grpc"
 )
@@ -164,6 +166,21 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 		}
 	}
 
+	// per-cluster HTTPS clients for volume server connections
+	var httpClientA, httpClientB *util_http_client.HTTPClient
+	if *syncOptions.aSecurity != "" {
+		var err error
+		if httpClientA, err = security.LoadHTTPClientFromFile(*syncOptions.aSecurity); err != nil {
+			glog.Fatalf("load HTTPS client config for filer A: %v", err)
+		}
+	}
+	if *syncOptions.bSecurity != "" {
+		var err error
+		if httpClientB, err = security.LoadHTTPClientFromFile(*syncOptions.bSecurity); err != nil {
+			glog.Fatalf("load HTTPS client config for filer B: %v", err)
+		}
+	}
+
 	grace.SetupProfiling(*syncCpuProfile, *syncMemProfile)
 
 	filerA := pb.ServerAddress(*syncOptions.filerA)
@@ -238,7 +255,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 				*syncOptions.bDoDeleteFiles,
 				aFilerSignature,
 				bFilerSignature,
-				&syncStateA2B)
+				&syncStateA2B,
+				httpClientA,
+				httpClientB)
 			if err != nil {
 				glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err)
 				time.Sleep(1747 * time.Millisecond)
@@ -279,7 +298,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 					*syncOptions.aDoDeleteFiles,
 					bFilerSignature,
 					aFilerSignature,
-					&syncStateB2A)
+					&syncStateB2A,
+					httpClientB,
+					httpClientA)
 				if err != nil {
 					glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err)
 					time.Sleep(2147 * time.Millisecond)
@@ -308,7 +329,8 @@ func initOffsetFromTsMs(grpcDialOption grpc.DialOption, targetFiler pb.ServerAdd
 }
 
 func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sourcePath string, sourceExcludePaths []string, sourceReadChunkFromFiler bool, targetGrpcDialOption grpc.DialOption, targetFiler pb.ServerAddress, targetPath string,
-	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, concurrency int, chunkConcurrency int, doDeleteFiles bool, sourceFilerSignature int32, targetFilerSignature int32, statePtr *atomic.Pointer[syncState]) error {
+	replicationStr, collection string, ttlSec int, sinkWriteChunkByFiler bool, diskType string, debug bool, concurrency int, chunkConcurrency int, doDeleteFiles bool, sourceFilerSignature int32, targetFilerSignature int32, statePtr *atomic.Pointer[syncState],
+	sourceHttpClient *util_http_client.HTTPClient, sinkHttpClient *util_http_client.HTTPClient) error {
 
 	// if first time, start from now
 	// if has previously synced, resume from that point of time
@@ -323,9 +345,15 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 	filerSource := &source.FilerSource{}
 	filerSource.DoInitialize(sourceFiler.ToHttpAddress(), sourceFiler.ToGrpcAddress(), sourcePath, sourceReadChunkFromFiler)
 	filerSource.SetGrpcDialOption(sourceGrpcDialOption)
+	if sourceHttpClient != nil {
+		filerSource.SetHttpClient(sourceHttpClient)
+	}
 	filerSink := &filersink.FilerSink{}
 	filerSink.DoInitialize(targetFiler.ToHttpAddress(), targetFiler.ToGrpcAddress(), targetPath, replicationStr, collection, ttlSec, diskType, targetGrpcDialOption, sinkWriteChunkByFiler)
 	filerSink.SetChunkConcurrency(chunkConcurrency)
+	if sinkHttpClient != nil {
+		filerSink.SetUploader(operation.NewUploaderWithHttpClient(sinkHttpClient))
+	}
 	filerSink.SetSourceFiler(filerSource)
 
 	persistEventFn := genProcessFunction(sourcePath, targetPath, sourceExcludePaths, nil, nil, nil, filerSink, doDeleteFiles, debug)

weed/operation/upload_content.go

@@ -135,6 +135,15 @@ func newUploader(httpClient HTTPClient) *Uploader {
 	}
 }
 
+// NewUploaderWithHttpClient creates an Uploader that uses the provided HTTP
+// client instead of the global one. This is used by filer.sync to upload to
+// remote clusters that use different TLS certificates.
+func NewUploaderWithHttpClient(httpClient HTTPClient) *Uploader {
+	return &Uploader{
+		httpClient: httpClient,
+	}
+}
+
 func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, host string, auth security.EncodedJwt, err error), uploadOption *UploadOption, genFileUrlFn func(host, fileId string) string, data []byte) (fileId string, uploadResult *UploadResult, err error) {
 	doUploadFunc := func() error {
 		var host string

weed/replication/sink/filersink/fetch_write.go

@@ -178,7 +178,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 }
 
 func (fs *FilerSink) uploadManifestChunk(path string, sourceMtime int64, sourceFileId string, manifestData []byte) (fileId string, err error) {
-	uploader, err := operation.NewUploader()
+	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload manifest data %v: %v", sourceFileId, err)
 		return "", fmt.Errorf("upload manifest data: %w", err)
@@ -235,7 +235,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtime int64, sourceF
 }
 
 func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtime int64) (fileId string, err error) {
-	uploader, err := operation.NewUploader()
+	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err)
 		return "", fmt.Errorf("upload data: %w", err)

weed/replication/sink/filersink/filer_sink.go

@@ -6,6 +6,7 @@ import (
 	"math"
 	"sync"
 
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 
@@ -50,6 +51,7 @@ type FilerSink struct {
 	executor          *util.LimitedConcurrentExecutor
 	signature         int32
 	activeTransfers   sync.Map // chunkFileId -> *ChunkTransferStatus
+	uploader          *operation.Uploader
 }
 
 func init() {
@@ -88,6 +90,21 @@ func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
 	fs.filerSource = s
 }
 
+// SetUploader sets a custom uploader for this sink, used when the target
+// cluster requires different TLS certificates than the global config.
+// Must be called during initialization, before any replication goroutines
+// start, since it writes fs.uploader without synchronization.
+func (fs *FilerSink) SetUploader(uploader *operation.Uploader) {
+	fs.uploader = uploader
+}
+
+func (fs *FilerSink) getUploader() (*operation.Uploader, error) {
+	if fs.uploader != nil {
+		return fs.uploader, nil
+	}
+	return operation.NewUploader()
+}
+
 func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
 	replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
 	fs.address = address

weed/replication/source/filer_source.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 )
 
 type FilerSource struct {
@@ -25,6 +26,7 @@ type FilerSource struct {
 	proxyByFiler   bool
 	dataCenter     string
 	signature      int32
+	httpClient     *util_http_client.HTTPClient
 }
 
 func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
@@ -54,6 +56,10 @@ func (fs *FilerSource) SetGrpcDialOption(option grpc.DialOption) {
 	fs.grpcDialOption = option
 }
 
+func (fs *FilerSource) SetHttpClient(client *util_http_client.HTTPClient) {
+	fs.httpClient = client
+}
+
 func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls []string, err error) {
 
 	vid2Locations := make(map[string]*filer_pb.Locations)
@@ -104,9 +110,15 @@ func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls
 }
 
 func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, header http.Header, resp *http.Response, err error) {
+	downloadFn := util_http.DownloadFile
+	if fs.httpClient != nil {
+		downloadFn = func(fileUrl string, jwt string, offset ...int64) (string, http.Header, *http.Response, error) {
+			return util_http.DownloadFileWithClient(fs.httpClient, fileUrl, jwt, offset...)
+		}
+	}
 
 	if fs.proxyByFiler {
-		filename, header, resp, err = util_http.DownloadFile("http://"+fs.address+"/?proxyChunkId="+fileId, "", offset)
+		filename, header, resp, err = downloadFn("http://"+fs.address+"/?proxyChunkId="+fileId, "", offset)
 		if err != nil {
 			glog.V(0).Infof("read part %s via filer proxy %s offset %d: %v", fileId, fs.address, offset, err)
 		} else {
@@ -121,7 +133,7 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h
 	}
 
 	for _, fileUrl := range fileUrls {
-		filename, header, resp, err = util_http.DownloadFile(fileUrl, "", offset)
+		filename, header, resp, err = downloadFn(fileUrl, "", offset)
 		if err != nil {
 			glog.V(0).Infof("fail to read part %s from %s offset %d: %v", fileId, fileUrl, offset, err)
 		} else {

weed/security/tls.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -209,6 +210,39 @@ func LoadClientTLS(config *util.ViperProxy, component string) grpc.DialOption {
 	return grpc.WithTransportCredentials(wrapped)
 }
 
+// LoadHTTPClientFromFile creates an HTTP client using the https.client TLS
+// settings from the given security config file. Returns nil if HTTPS is not
+// enabled in the config. This is used by filer.sync to create per-cluster
+// HTTP clients when clusters use different certificates.
+func LoadHTTPClientFromFile(configFile string) (*util_http_client.HTTPClient, error) {
+	v := viper.New()
+	v.SetConfigFile(configFile)
+	if err := v.ReadInConfig(); err != nil {
+		return nil, fmt.Errorf("failed to read security config %s: %v", configFile, err)
+	}
+
+	if !v.GetBool("https.client.enabled") {
+		return nil, nil
+	}
+
+	configDir := filepath.Dir(configFile)
+	resolvePath := func(key string) string {
+		p := v.GetString(key)
+		if p != "" && !filepath.IsAbs(p) {
+			return filepath.Join(configDir, p)
+		}
+		return p
+	}
+
+	return util_http_client.NewHttpClientWithTLS(
+		resolvePath("https.client.cert"),
+		resolvePath("https.client.key"),
+		resolvePath("https.client.ca"),
+		v.GetBool("https.client.insecure_skip_verify"),
+		util_http_client.AddDialContext,
+	)
+}
+
 func LoadClientTLSHTTP(clientCertFile string) *tls.Config {
 	clientCerts, err := os.ReadFile(clientCertFile)
 	if err != nil {

weed/util/http/client/http_client.go

@@ -195,6 +195,66 @@ func getClientCaCert(clientName ClientName) ([]byte, string, error) {
 	return getFileContentFromSecurityConfiguration(clientName, "ca")
 }
 
+// NewHttpClientWithTLS creates an HTTPClient with explicit TLS certificate
+// parameters instead of reading from the global security configuration.
+// This is used by filer.sync to create per-cluster HTTP clients when clusters
+// use different certificates.
+func NewHttpClientWithTLS(certFile, keyFile, caFile string, insecureSkipVerify bool, opts ...HttpClientOpt) (*HTTPClient, error) {
+	httpClient := HTTPClient{}
+	httpClient.expectHttpsScheme = true
+	var tlsConfig *tls.Config
+
+	if (certFile == "") != (keyFile == "") {
+		return nil, fmt.Errorf("both cert and key are required for mTLS, got cert=%q key=%q", certFile, keyFile)
+	}
+
+	var clientCert *tls.Certificate
+	if certFile != "" && keyFile != "" {
+		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
+		if err != nil {
+			return nil, fmt.Errorf("error loading client certificate and key: %s", err)
+		}
+		clientCert = &cert
+	}
+
+	var caCertPool *x509.CertPool
+	if caFile != "" {
+		caCert, err := os.ReadFile(caFile)
+		if err != nil {
+			return nil, fmt.Errorf("error reading CA cert %s: %s", caFile, err)
+		}
+		caCertPool, err = createHTTPClientCertPool(caCert, caFile)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if clientCert != nil || caCertPool != nil || insecureSkipVerify {
+		tlsConfig = &tls.Config{
+			Certificates:       []tls.Certificate{},
+			RootCAs:            caCertPool,
+			InsecureSkipVerify: insecureSkipVerify,
+		}
+		if clientCert != nil {
+			tlsConfig.Certificates = append(tlsConfig.Certificates, *clientCert)
+		}
+	}
+
+	httpClient.Transport = &http.Transport{
+		MaxIdleConns:        1024,
+		MaxIdleConnsPerHost: 1024,
+		TLSClientConfig:     tlsConfig,
+	}
+	httpClient.Client = &http.Client{
+		Transport: httpClient.Transport,
+	}
+
+	for _, opt := range opts {
+		opt(&httpClient)
+	}
+	return &httpClient, nil
+}
+
 func createHTTPClientCertPool(certContent []byte, fileName string) (*x509.CertPool, error) {
 	certPool := x509.NewCertPool()
 	if len(certContent) == 0 {

weed/util/http/http_global_client_util.go

@@ -21,6 +21,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 
 	"github.com/seaweedfs/seaweedfs/weed/security"
+	util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client"
 )
 
 var ErrNotFound = fmt.Errorf("not found")
@@ -202,6 +203,16 @@ func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) e
 }
 
 func DownloadFile(fileUrl string, jwt string, offset ...int64) (filename string, header http.Header, resp *http.Response, e error) {
+	return DownloadFileWithClient(GetGlobalHttpClient(), fileUrl, jwt, offset...)
+}
+
+// DownloadFileWithClient is like DownloadFile but uses the provided HTTP client
+// instead of the global one. This is used by filer.sync to download from
+// remote clusters that use different TLS certificates.
+func DownloadFileWithClient(client *util_http_client.HTTPClient, fileUrl string, jwt string, offset ...int64) (filename string, header http.Header, resp *http.Response, e error) {
+	if client == nil {
+		return "", nil, nil, fmt.Errorf("nil HTTP client in DownloadFileWithClient")
+	}
 	req, err := http.NewRequest(http.MethodGet, fileUrl, nil)
 	if err != nil {
 		return "", nil, nil, err
@@ -217,7 +228,7 @@ func DownloadFile(fileUrl string, jwt string, offset ...int64) (filename string,
 		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", rangeOffset))
 	}
 
-	response, err := GetGlobalHttpClient().Do(req)
+	response, err := client.Do(req)
 	if err != nil {
 		return "", nil, nil, err
 	}