From 2919bb27e529fb0377678df023fc3d2f1609934b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 7 Apr 2026 14:11:44 -0700 Subject: [PATCH] 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 | 34 ++++++++++- weed/operation/upload_content.go | 9 +++ .../replication/sink/filersink/fetch_write.go | 4 +- weed/replication/sink/filersink/filer_sink.go | 17 ++++++ weed/replication/source/filer_source.go | 16 ++++- weed/security/tls.go | 34 +++++++++++ weed/util/http/client/http_client.go | 60 +++++++++++++++++++ weed/util/http/http_global_client_util.go | 13 +++- 8 files changed, 179 insertions(+), 8 deletions(-) diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index a509deb48..ca74d3aec 100644 --- a/weed/command/filer_sync.go +++ b/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) diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 9e7b6c4e8..47ae230c5 100644 --- a/weed/operation/upload_content.go +++ b/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 diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index a2a12f95a..0399542f1 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/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) diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index a51d1c050..50aaab449 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/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 diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index 32c2a2235..ad8683488 100644 --- a/weed/replication/source/filer_source.go +++ b/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 { diff --git a/weed/security/tls.go b/weed/security/tls.go index 78d4e501a..2e2af94b3 100644 --- a/weed/security/tls.go +++ b/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 { diff --git a/weed/util/http/client/http_client.go b/weed/util/http/client/http_client.go index ca908763f..3291c56a2 100644 --- a/weed/util/http/client/http_client.go +++ b/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 { diff --git a/weed/util/http/http_global_client_util.go b/weed/util/http/http_global_client_util.go index f8d20ae67..1c8af2225 100644 --- a/weed/util/http/http_global_client_util.go +++ b/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 }