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") + } +}