From 78893d274ae4d53d3be4370c2d2925bb32c0ca4f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 7 Oct 2026 07:28:17 -0700 Subject: [PATCH] filer.sync: sign proxied chunk I/O from the per-side security file The -a.security / -b.security files were used for gRPC TLS and the HTTPS client but not for jwt.filer_signing, so filer-proxied chunk reads and writes carried a token signed with the process-wide key and failed authorization whenever the two clusters' keys differ. LoadFilerJwtFromFile returns a FilerJwtProvider for each side's file, which FilerSource and FilerSink now accept for proxied chunk reads and writes. With no keys in the file or no flag, both fall back to the process-wide jwt.filer_signing configuration as before. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/filer_sync.go | 24 +++++- weed/operation/upload_content.go | 7 +- .../replication/sink/filersink/fetch_write.go | 2 + weed/replication/sink/filersink/filer_sink.go | 7 ++ weed/replication/source/filer_source.go | 11 ++- weed/security/jwt.go | 36 +++++++++ weed/security/jwt_filer_provider_test.go | 77 +++++++++++++++++++ 7 files changed, 159 insertions(+), 5 deletions(-) create mode 100644 weed/security/jwt_filer_provider_test.go diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index d987f57999..c71c5b16ab 100644 --- a/weed/command/filer_sync.go +++ b/weed/command/filer_sync.go @@ -174,17 +174,24 @@ func runFilerSynchronize(cmd *Command, args []string) bool { // per-cluster HTTPS clients for volume server connections var httpClientA, httpClientB *util_http_client.HTTPClient + var jwtForFilerA, jwtForFilerB security.FilerJwtProvider 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 jwtForFilerA, err = security.LoadFilerJwtFromFile(*syncOptions.aSecurity); err != nil { + glog.Fatalf("load filer JWT 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) } + if jwtForFilerB, err = security.LoadFilerJwtFromFile(*syncOptions.bSecurity); err != nil { + glog.Fatalf("load filer JWT config for filer B: %v", err) + } } grace.SetupProfiling(*syncCpuProfile, *syncMemProfile) @@ -263,7 +270,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool { bFilerSignature, &syncStateA2B, httpClientA, - httpClientB) + httpClientB, + jwtForFilerA, + jwtForFilerB) if err != nil { glog.Errorf("sync from %s to %s: %v", *syncOptions.filerA, *syncOptions.filerB, err) time.Sleep(1747 * time.Millisecond) @@ -306,7 +315,9 @@ func runFilerSynchronize(cmd *Command, args []string) bool { aFilerSignature, &syncStateB2A, httpClientB, - httpClientA) + httpClientA, + jwtForFilerB, + jwtForFilerA) if err != nil { glog.Errorf("sync from %s to %s: %v", *syncOptions.filerB, *syncOptions.filerA, err) time.Sleep(2147 * time.Millisecond) @@ -336,7 +347,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], - sourceHttpClient *util_http_client.HTTPClient, sinkHttpClient *util_http_client.HTTPClient) error { + sourceHttpClient *util_http_client.HTTPClient, sinkHttpClient *util_http_client.HTTPClient, + sourceJwtProvider security.FilerJwtProvider, sinkJwtProvider security.FilerJwtProvider) error { // if first time, start from now // if has previously synced, resume from that point of time @@ -357,12 +369,18 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi if sourceHttpClient != nil { filerSource.SetHttpClient(sourceHttpClient) } + if sourceJwtProvider != nil { + filerSource.SetFilerJwtProvider(sourceJwtProvider) + } 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)) } + if sinkJwtProvider != nil { + filerSink.SetFilerJwtProvider(sinkJwtProvider) + } 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 4ce3e61491..288b6b9ee8 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -62,6 +62,7 @@ type UploadOption struct { SourceUrl string // optional: for logging when reading from a remote source MaxAttempts int // <=0 uses the default GenUploadUrl func(host, fileId string) string // if nil → fallback "http://{host}/{fileId}" + FilerJwt security.FilerJwtProvider // credential for proxy chunk URLs; nil → process-wide jwt.filer_signing } type UploadResult struct { @@ -217,7 +218,11 @@ func (uploader *Uploader) uploadWithRetryData(assignFn func() (fileId string, ho // The request addresses the filer, which authorizes it and mints the // volume credential itself. The AssignVolume token is not a filer // credential and gets the caller nowhere here. - uploadOption.Jwt = security.EncodedJwt(util_http.JwtForFilerServer(true)) + if uploadOption.FilerJwt != nil { + uploadOption.Jwt = uploadOption.FilerJwt(true) + } else { + uploadOption.Jwt = security.EncodedJwt(util_http.JwtForFilerServer(true)) + } } uploadResult, err = uploader.retriedUploadData(context.Background(), data, uploadOption) diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 08a79bffcb..809ba1100c 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -315,6 +315,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc } if fs.writeChunkByFiler { uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address) + uploadOption.FilerJwt = fs.jwtForFiler } currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry( fs, @@ -435,6 +436,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, } if fs.writeChunkByFiler { uploadOption.GenUploadUrl = operation.GenUploadUrlProxy(fs.address) + uploadOption.FilerJwt = fs.jwtForFiler } currentFileId, uploadResult, uploadErr, _ := uploader.UploadWithRetry( fs, diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 868cb01fc1..3c278d9e54 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -62,6 +62,7 @@ type FilerSink struct { signature int32 activeTransfers sync.Map // chunkFileId -> *ChunkTransferStatus uploader *operation.Uploader + jwtForFiler security.FilerJwtProvider // lastServedFileId is the most recent chunk the source did serve, the probe // sourceStillServesChunks re-checks before writing an entry off. lastServedFileId atomic.Pointer[string] @@ -116,6 +117,12 @@ func (fs *FilerSink) SetUploader(uploader *operation.Uploader) { fs.uploader = uploader } +// SetFilerJwtProvider sets the filer API credential for proxied chunk writes. +// Must be called during initialization, before any replication goroutines start. +func (fs *FilerSink) SetFilerJwtProvider(provider security.FilerJwtProvider) { + fs.jwtForFiler = provider +} + func (fs *FilerSink) getUploader() (*operation.Uploader, error) { if fs.uploader != nil { return fs.uploader, nil diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index 9b834b0fa5..945781d915 100644 --- a/weed/replication/source/filer_source.go +++ b/weed/replication/source/filer_source.go @@ -34,6 +34,7 @@ type FilerSource struct { dataCenter string signature int32 httpClient *util_http_client.HTTPClient + jwtForFiler security.FilerJwtProvider } func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error { @@ -67,6 +68,10 @@ func (fs *FilerSource) SetHttpClient(client *util_http_client.HTTPClient) { fs.httpClient = client } +func (fs *FilerSource) SetFilerJwtProvider(provider security.FilerJwtProvider) { + fs.jwtForFiler = provider +} + func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls []string, err error) { vid2Locations := make(map[string]*filer_pb.Locations) @@ -126,7 +131,11 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h if fs.proxyByFiler { fileUrl := util_http.ProxyChunkUrl(fs.address, fileId) - filename, header, resp, err = downloadFn(fileUrl, util_http.JwtForFilerServer(false), offset) + jwt := util_http.JwtForFilerServer(false) + if fs.jwtForFiler != nil { + jwt = string(fs.jwtForFiler(false)) + } + filename, header, resp, err = downloadFn(fileUrl, jwt, offset) if err == nil { err = readPartStatusError(fileUrl, resp) } diff --git a/weed/security/jwt.go b/weed/security/jwt.go index 47f15e7843..780beea324 100644 --- a/weed/security/jwt.go +++ b/weed/security/jwt.go @@ -8,6 +8,7 @@ import ( jwt "github.com/golang-jwt/jwt/v5" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/spf13/viper" ) type EncodedJwt string @@ -151,3 +152,38 @@ func DecodeJwt(signingKey SigningKey, tokenString EncodedJwt, claims jwt.Claims) return []byte(signingKey), nil }) } + +// FilerJwtProvider signs the credential a filer's HTTP API expects. A nil +// provider means the process-wide jwt.filer_signing configuration applies. +type FilerJwtProvider func(isWrite bool) EncodedJwt + +// LoadFilerJwtFromFile reads jwt.filer_signing from a security file the way +// LoadClientTLSFromFile reads the TLS section. A nil provider means the file +// configures no filer signing keys and the process-wide configuration applies. +func LoadFilerJwtFromFile(configFile string) (FilerJwtProvider, 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) + } + + signingKey := SigningKey(v.GetString("jwt.filer_signing.key")) + readSigningKey := SigningKey(v.GetString("jwt.filer_signing.read.key")) + if len(signingKey) == 0 && len(readSigningKey) == 0 { + return nil, nil + } + signingKeyExpires := v.GetInt("jwt.filer_signing.expires_after_seconds") + if signingKeyExpires == 0 { + signingKeyExpires = 10 + } + readSigningKeyExpires := v.GetInt("jwt.filer_signing.read.expires_after_seconds") + if readSigningKeyExpires == 0 { + readSigningKeyExpires = 60 + } + return func(isWrite bool) EncodedJwt { + if isWrite { + return GenJwtForFilerServer(signingKey, signingKeyExpires) + } + return GenJwtForFilerServer(readSigningKey, readSigningKeyExpires) + }, nil +} diff --git a/weed/security/jwt_filer_provider_test.go b/weed/security/jwt_filer_provider_test.go new file mode 100644 index 0000000000..c3162dace6 --- /dev/null +++ b/weed/security/jwt_filer_provider_test.go @@ -0,0 +1,77 @@ +package security + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadFilerJwtFromFile(t *testing.T) { + configFile := filepath.Join(t.TempDir(), "security.toml") + config := ` +[jwt] +[jwt.filer_signing] +key = "side-write-key" +expires_after_seconds = 30 +[jwt.filer_signing.read] +key = "side-read-key" +expires_after_seconds = 90 +` + if err := os.WriteFile(configFile, []byte(config), 0644); err != nil { + t.Fatal(err) + } + + provider, err := LoadFilerJwtFromFile(configFile) + if err != nil { + t.Fatal(err) + } + if provider == nil { + t.Fatal("config with filer signing keys gave a nil provider") + } + + for _, tc := range []struct { + name string + isWrite bool + signedBy string + otherKey string + expires int64 + }{ + {"read", false, "side-read-key", "side-write-key", 90}, + {"write", true, "side-write-key", "side-read-key", 30}, + } { + t.Run(tc.name, func(t *testing.T) { + token := provider(tc.isWrite) + claims := &SeaweedFilerClaims{} + if _, err := DecodeJwt(SigningKey(tc.signedBy), token, claims); err != nil { + t.Fatalf("token does not validate against the %s key: %v", tc.name, err) + } + if claims.ExpiresAt == nil { + t.Fatal("token never expires") + } + if _, err := DecodeJwt(SigningKey(tc.otherKey), token, &SeaweedFilerClaims{}); err == nil { + t.Fatal("token also validates against the other access level's key") + } + }) + } +} + +func TestLoadFilerJwtFromFileWithoutKeys(t *testing.T) { + configFile := filepath.Join(t.TempDir(), "security.toml") + if err := os.WriteFile(configFile, []byte("[grpc.client]\n"), 0644); err != nil { + t.Fatal(err) + } + + provider, err := LoadFilerJwtFromFile(configFile) + if err != nil { + t.Fatal(err) + } + if provider != nil { + t.Fatal("config without filer signing keys gave a provider") + } +} + +func TestLoadFilerJwtFromFileMissing(t *testing.T) { + if _, err := LoadFilerJwtFromFile(filepath.Join(t.TempDir(), "none.toml")); err == nil { + t.Fatal("missing config file loaded without error") + } +}