diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index d987f5799..c71c5b16a 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/filer/filechunk_group.go b/weed/filer/filechunk_group.go index 2a8d1894f..0c6f94374 100644 --- a/weed/filer/filechunk_group.go +++ b/weed/filer/filechunk_group.go @@ -240,7 +240,7 @@ func (group *ChunkGroup) SetChunks(chunks []*filer_pb.FileChunk) error { continue } - resolvedChunks, err := resolveOneChunkManifest(context.Background(), group.lookupFn, chunk, group.cacheInvalidator, group.manifestCache) + resolvedChunks, err := resolveOneChunkManifest(context.Background(), group.lookupFn, chunk, group.cacheInvalidator, group.manifestCache, nil) if err != nil { group.resolveErr = err return err diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index f800333c9..a8661549c 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -16,6 +16,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) @@ -55,7 +56,16 @@ func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonMa } func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64, invalidator CacheInvalidator) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { - resolver := newChunkManifestResolver(ctx, lookupFileIdFn, invalidator) + resolver := newChunkManifestResolver(ctx, lookupFileIdFn, invalidator, nil) + defer resolver.close() + return resolver.resolve(chunks, startOffset, stopOffset) +} + +// ResolveChunkManifestWithFilerJwt is ResolveChunkManifest that signs proxied +// manifest downloads with filerJwtFn instead of the process-wide filer read +// key, for readers carrying per-source credentials. +func ResolveChunkManifestWithFilerJwt(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, filerJwtFn security.FilerJwtProvider, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64, invalidator CacheInvalidator) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { + resolver := newChunkManifestResolver(ctx, lookupFileIdFn, invalidator, filerJwtFn) defer resolver.close() return resolver.resolve(chunks, startOffset, stopOffset) } @@ -81,6 +91,7 @@ type chunkManifestResolver struct { cancel context.CancelFunc lookupFileIdFn wdclient.LookupFileIdFunctionType invalidator CacheInvalidator + filerJwtFn security.FilerJwtProvider jobs chan chunkManifestResolveJob overflowSem chan struct{} workers sync.WaitGroup @@ -88,7 +99,7 @@ type chunkManifestResolver struct { started bool } -func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, invalidator CacheInvalidator) *chunkManifestResolver { +func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, invalidator CacheInvalidator, filerJwtFn security.FilerJwtProvider) *chunkManifestResolver { workCtx, cancel := context.WithCancel(ctx) resolver := &chunkManifestResolver{ ctx: workCtx, @@ -96,6 +107,7 @@ func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.Looku cancel: cancel, lookupFileIdFn: lookupFileIdFn, invalidator: invalidator, + filerJwtFn: filerJwtFn, jobs: make(chan chunkManifestResolveJob, chunkManifestResolveJobBufferSize), overflowSem: make(chan struct{}, maxChunkManifestResolveWorkers), } @@ -103,7 +115,7 @@ func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.Looku } func (r *chunkManifestResolver) executeJob(job chunkManifestResolveJob) { - job.result.chunks, job.result.err = ResolveOneChunkManifest(job.batchCtx, r.lookupFileIdFn, job.chunk, r.invalidator) + job.result.chunks, job.result.err = resolveOneChunkManifest(job.batchCtx, r.lookupFileIdFn, job.chunk, r.invalidator, nil, r.filerJwtFn) if job.result.err != nil && r.parentCtx.Err() == nil { if job.batchCtx.Err() != nil && errors.Is(job.result.err, context.Canceled) { job.result.internalCancel = true @@ -293,14 +305,23 @@ func (r *chunkManifestResolver) resolve(chunks []*filer_pb.FileChunk, startOffse // Keeping this signature stable preserves the existing four-argument contract // for external callers. func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { - return resolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator, nil) + return resolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator, nil, nil) +} + +// ResolveOneChunkManifestWithFilerJwt is ResolveOneChunkManifest that signs +// proxied manifest downloads with filerJwtFn instead of the process-wide +// filer read key, for readers carrying per-source credentials. +func ResolveOneChunkManifestWithFilerJwt(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, filerJwtFn security.FilerJwtProvider, chunk *filer_pb.FileChunk, invalidator CacheInvalidator) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { + return resolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator, nil, filerJwtFn) } // resolveOneChunkManifest is the cache-aware implementation. cache may be nil, // in which case the manifest is fetched and validated on every call, matching // the historical uncached behavior. A non-nil cache is owned by a single mount -// (WFS) and coalesces concurrent cold misses via singleflight. -func resolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator, cache *ChunkManifestCache) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { +// (WFS) and coalesces concurrent cold misses via singleflight. filerJwtFn +// overrides the filer read credential for proxied downloads; nil means the +// process-wide key. +func resolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator, cache *ChunkManifestCache, filerJwtFn security.FilerJwtProvider) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { if !chunk.IsChunkManifest { return } @@ -318,7 +339,7 @@ func resolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.Lookup bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer) bytesBuffer.Reset() defer bytesBufferPool.Put(bytesBuffer) - if err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, key.fileID, chunk.CipherKey, chunk.IsCompressed, invalidator); err != nil { + if err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, key.fileID, chunk.CipherKey, chunk.IsCompressed, invalidator, filerJwtFn); err != nil { return nil, fmt.Errorf("fail to read manifest %s: %w", key.fileID, err) } // Copy before the buffer returns to the pool so concurrent callers @@ -357,13 +378,18 @@ func resolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.Lookup return m.Chunks, nil } -func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, invalidator CacheInvalidator) error { +func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, invalidator CacheInvalidator, filerJwtFn security.FilerJwtProvider) error { urlStrings, err := lookupFileIdFn(ctx, fileId) if err != nil { glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", fileId, err) return err } - jwt := ChunkReadJwt(urlStrings, fileId) + var jwt string + if filerJwtFn != nil && len(urlStrings) > 0 && util_http.IsProxyChunkUrl(urlStrings[0]) { + jwt = string(filerJwtFn(false)) + } else { + jwt = ChunkReadJwt(urlStrings, fileId) + } if _, err = retriedStreamFetchChunkData(ctx, bytesBuffer, urlStrings, jwt, cipherKey, isGzipped, true, 0, 0, refreshUrls(ctx, invalidator, lookupFileIdFn, fileId)); err == nil { return nil } diff --git a/weed/filer/filechunk_manifest_cache_benchmark_test.go b/weed/filer/filechunk_manifest_cache_benchmark_test.go index 91903d925..75b6a1449 100644 --- a/weed/filer/filechunk_manifest_cache_benchmark_test.go +++ b/weed/filer/filechunk_manifest_cache_benchmark_test.go @@ -23,7 +23,7 @@ func BenchmarkManifestResolutionRepeatedOpen(b *testing.B) { } chunk := newManifestCacheTestChunk("benchmark-cached") cache := NewChunkManifestCache(MaxMountChunkManifestCacheEntries, MaxMountChunkManifestCacheBytes) - _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache) + _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache, nil) if err != nil { b.Fatal(err) } @@ -33,7 +33,7 @@ func BenchmarkManifestResolutionRepeatedOpen(b *testing.B) { b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { - if _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache); err != nil { + if _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache, nil); err != nil { b.Fatal(err) } } diff --git a/weed/filer/filechunk_manifest_cache_test.go b/weed/filer/filechunk_manifest_cache_test.go index af0ffd34c..21644ce76 100644 --- a/weed/filer/filechunk_manifest_cache_test.go +++ b/weed/filer/filechunk_manifest_cache_test.go @@ -258,7 +258,7 @@ func TestResolveOneChunkManifestHonorsCanceledContextOnCacheHit(t *testing.T) { return nil, errors.New("lookup should not be called") } - _, err := resolveOneChunkManifest(ctx, lookup, chunk, nil, cache) + _, err := resolveOneChunkManifest(ctx, lookup, chunk, nil, cache, nil) require.ErrorIs(t, err, context.Canceled) require.False(t, lookupCalled, "a canceled cache hit must not issue a lookup") } @@ -278,7 +278,7 @@ func TestResolveOneChunkManifestCanceledWaiterReturnsDuringCoalescedMiss(t *test defer leaderCancel() leaderDone := make(chan error, 1) go func() { - _, err := resolveOneChunkManifest(leaderCtx, fixture.lookup, chunk, nil, cache) + _, err := resolveOneChunkManifest(leaderCtx, fixture.lookup, chunk, nil, cache, nil) leaderDone <- err }() @@ -287,7 +287,7 @@ func TestResolveOneChunkManifestCanceledWaiterReturnsDuringCoalescedMiss(t *test // for the leader's result. waiterCtx, waiterCancel := context.WithCancel(context.Background()) waiterCancel() - _, err := resolveOneChunkManifest(waiterCtx, fixture.lookup, chunk, nil, cache) + _, err := resolveOneChunkManifest(waiterCtx, fixture.lookup, chunk, nil, cache, nil) require.ErrorIs(t, err, context.Canceled) // The leader must still complete successfully and populate the cache. diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 2036a8d11..2d5d9f412 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -16,6 +16,8 @@ import ( "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) func TestDoMaybeManifestize(t *testing.T) { @@ -621,7 +623,7 @@ func TestFetchWholeChunkRetriesFreshLocations(t *testing.T) { inv := &countingInvalidator{} bytesBuffer := fetchManifestBuffer(t) - assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,stale", nil, false, inv)) + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,stale", nil, false, inv, nil)) assert.Equal(t, int32(1), inv.invalidations.Load()) assert.Equal(t, int32(2), lookup.calls.Load()) @@ -655,7 +657,7 @@ func TestFetchWholeChunkRefreshesLocationsAfterPartialFailure(t *testing.T) { inv := &countingInvalidator{} bytesBuffer := fetchManifestBuffer(t) - assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,abc", nil, false, inv)) + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,abc", nil, false, inv, nil)) assert.Equal(t, int32(1), inv.invalidations.Load()) assert.Equal(t, int32(2), lookup.calls.Load()) decoded := &filer_pb.FileChunkManifest{} @@ -676,7 +678,7 @@ func TestFetchWholeChunkWithoutInvalidator(t *testing.T) { freshUrls: []string{"http://unused:8080/5,abc"}, } - assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, nil)) + assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, nil, nil)) assert.Equal(t, int32(1), lookup.calls.Load()) } @@ -694,7 +696,7 @@ func TestFetchWholeChunkUnchangedLocations(t *testing.T) { } inv := &countingInvalidator{} - assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv)) + assert.Error(t, fetchWholeChunk(context.Background(), fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv, nil)) assert.Equal(t, int32(2), lookup.calls.Load()) assert.Equal(t, int32(1), inv.invalidations.Load()) } @@ -711,7 +713,7 @@ func TestFetchWholeChunkCancelledKeepsLocations(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - err := fetchWholeChunk(ctx, fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv) + err := fetchWholeChunk(ctx, fetchManifestBuffer(t), lookup.lookup, "5,abc", nil, false, inv, nil) assert.ErrorIs(t, err, context.Canceled) assert.Equal(t, int32(0), inv.invalidations.Load()) assert.Equal(t, int32(1), lookup.calls.Load()) @@ -730,3 +732,43 @@ func TestFetchWholeChunkCancelledKeepsLocations(t *testing.T) { }) assert.ErrorIs(t, noInvalidator, context.Canceled) } + +// TestFetchWholeChunkUsesProvidedFilerJwt covers a replicating reader whose +// source filer authenticates proxied downloads with its own read key: the +// supplied provider's token must reach the server, not the process-wide one. +func TestFetchWholeChunkUsesProvidedFilerJwt(t *testing.T) { + manifestBytes, err := proto.Marshal(&filer_pb.FileChunkManifest{ + Chunks: []*filer_pb.FileChunk{{FileId: "100,abc", Offset: 0, Size: 8}}, + }) + assert.NoError(t, err) + + gotAuth := make(chan string, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth <- r.Header.Get("Authorization") + w.Header().Set("Content-Length", strconv.Itoa(len(manifestBytes))) + w.Write(manifestBytes) + })) + t.Cleanup(srv.Close) + + lookup := func(ctx context.Context, fileId string) ([]string, error) { + return []string{srv.URL + "/?" + util_http.ProxyChunkIdParam + "=" + fileId}, nil + } + jwtFn := func(isWrite bool) security.EncodedJwt { + assert.False(t, isWrite) + return "side-read-jwt" + } + bytesBuffer := fetchManifestBuffer(t) + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup, "5,abc", nil, false, nil, jwtFn)) + assert.Equal(t, security.BearerPrefix+"side-read-jwt", <-gotAuth) + + // non-proxy URLs keep the volume-server credential and never call the provider + volumeURL := manifestServer(t, manifestBytes).URL + "/5,abc" + volumeLookup := func(ctx context.Context, fileId string) ([]string, error) { + return []string{volumeURL}, nil + } + bytesBuffer.Reset() + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, volumeLookup, "5,abc", nil, false, nil, func(bool) security.EncodedJwt { + t.Fatal("provider must not be consulted for a volume url") + return "" + })) +} diff --git a/weed/filer/persisted_log_cache.go b/weed/filer/persisted_log_cache.go index 430582c21..0250c8c00 100644 --- a/weed/filer/persisted_log_cache.go +++ b/weed/filer/persisted_log_cache.go @@ -190,7 +190,7 @@ func loadLogFileEntries(masterClient *wdclient.MasterClient, chunk *filer_pb.Fil lookupFileIdFn := func(ctx context.Context, fileId string) (targetUrls []string, err error) { return masterClient.LookupFileId(ctx, fileId) } - if fetchErr := fetchWholeChunk(context.Background(), bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed, masterClient); fetchErr != nil { + if fetchErr := fetchWholeChunk(context.Background(), bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed, masterClient, nil); fetchErr != nil { return nil, false, fetchErr } return decodeLogRecords(bytesBuffer.Bytes()) diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 4ce3e6149..288b6b9ee 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 08a79bffc..1c165c7d1 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -195,7 +195,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk resolveName := fmt.Sprintf("resolve manifest %s", sourceChunk.GetFileIdString()) missingGate := fs.newMissingSourceChunkGate(sourceChunk.GetFileIdString()) err := util.RetryUntil(resolveName, func() error { - rc, e := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk, nil) + rc, e := filer.ResolveOneChunkManifestWithFilerJwt(ctx, fs.filerSource.LookupFileId, fs.filerSource.FilerJwt(), sourceChunk, nil) if e != nil { return e } @@ -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 868cb01fc..34b68c768 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 @@ -358,7 +365,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent existingEntry.RemoteEntry = newEntry.RemoteEntry default: // source-side chunks resolve via source filer; sink volume IDs may collide. - deletedChunks, newChunks, err := compareChunks(context.Background(), filer.LookupFn(fs.filerSource), oldEntry, newEntry) + deletedChunks, newChunks, err := compareChunks(context.Background(), filer.LookupFn(fs.filerSource), fs.filerSource.FilerJwt(), oldEntry, newEntry) if err != nil { return true, fmt.Errorf("replicate %s compare chunks error: %w", key, err) } @@ -404,12 +411,12 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent }) } -func compareChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) { - aData, aMeta, aErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, oldEntry.GetChunks(), 0, math.MaxInt64, nil) +func compareChunks(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, filerJwtFn security.FilerJwtProvider, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) { + aData, aMeta, aErr := filer.ResolveChunkManifestWithFilerJwt(ctx, lookupFileIdFn, filerJwtFn, oldEntry.GetChunks(), 0, math.MaxInt64, nil) if aErr != nil { return nil, nil, aErr } - bData, bMeta, bErr := filer.ResolveChunkManifest(ctx, lookupFileIdFn, newEntry.GetChunks(), 0, math.MaxInt64, nil) + bData, bMeta, bErr := filer.ResolveChunkManifestWithFilerJwt(ctx, lookupFileIdFn, filerJwtFn, newEntry.GetChunks(), 0, math.MaxInt64, nil) if bErr != nil { return nil, nil, bErr } diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index 9b834b0fa..376a51804 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,16 @@ func (fs *FilerSource) SetHttpClient(client *util_http_client.HTTPClient) { fs.httpClient = client } +func (fs *FilerSource) SetFilerJwtProvider(provider security.FilerJwtProvider) { + fs.jwtForFiler = provider +} + +// FilerJwt returns the side-specific filer API credential, or nil when the +// process-wide jwt.filer_signing configuration applies. +func (fs *FilerSource) FilerJwt() security.FilerJwtProvider { + return fs.jwtForFiler +} + func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls []string, err error) { vid2Locations := make(map[string]*filer_pb.Locations) @@ -126,7 +137,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 47f15e784..30b02db3e 100644 --- a/weed/security/jwt.go +++ b/weed/security/jwt.go @@ -8,6 +8,8 @@ import ( jwt "github.com/golang-jwt/jwt/v5" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/spf13/viper" ) type EncodedJwt string @@ -151,3 +153,57 @@ 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, honoring the same WEED_ +// environment precedence. A nil provider means the file configures no filer +// signing keys and the process-wide configuration applies. A file that sets +// only one access level falls back to the process-wide key for the other. +func LoadFilerJwtFromFile(configFile string) (FilerJwtProvider, error) { + v := viper.New() + v.SetConfigFile(configFile) + v.AutomaticEnv() + v.SetEnvPrefix("weed") + v.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) + 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") + readSigningKeyExpires := v.GetInt("jwt.filer_signing.read.expires_after_seconds") + if len(signingKey) == 0 || len(readSigningKey) == 0 { + gv := util.GetViper() + if len(signingKey) == 0 { + signingKey = SigningKey(gv.GetString("jwt.filer_signing.key")) + signingKeyExpires = gv.GetInt("jwt.filer_signing.expires_after_seconds") + } + if len(readSigningKey) == 0 { + readSigningKey = SigningKey(gv.GetString("jwt.filer_signing.read.key")) + readSigningKeyExpires = gv.GetInt("jwt.filer_signing.read.expires_after_seconds") + } + } + if signingKeyExpires < 0 || readSigningKeyExpires < 0 { + return nil, fmt.Errorf("jwt.filer_signing lifetimes must not be negative") + } + if signingKeyExpires == 0 { + signingKeyExpires = 10 + } + 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 000000000..8a4a6067b --- /dev/null +++ b/weed/security/jwt_filer_provider_test.go @@ -0,0 +1,146 @@ +package security + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +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) { + before := time.Now() + token := provider(tc.isWrite) + after := time.Now() + 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") + } + expiresIn := claims.ExpiresAt.Time + if expiresIn.Before(before.Add(time.Duration(tc.expires-1)*time.Second)) || expiresIn.After(after.Add(time.Duration(tc.expires+1)*time.Second)) { + t.Fatalf("token expires at %v, want %ds after %v", expiresIn, tc.expires, before) + } + 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") + } +} + +func TestLoadFilerJwtFromFilePartialKeys(t *testing.T) { + configFile := filepath.Join(t.TempDir(), "security.toml") + config := ` +[jwt.filer_signing.read] +key = "side-read-key" +` + if err := os.WriteFile(configFile, []byte(config), 0644); err != nil { + t.Fatal(err) + } + + gv := util.GetViper() + priorKey := gv.GetString("jwt.filer_signing.key") + gv.Set("jwt.filer_signing.key", "global-write-key") + t.Cleanup(func() { gv.Set("jwt.filer_signing.key", priorKey) }) + + provider, err := LoadFilerJwtFromFile(configFile) + if err != nil { + t.Fatal(err) + } + if provider == nil { + t.Fatal("config with a filer signing key gave a nil provider") + } + + if _, err := DecodeJwt(SigningKey("global-write-key"), provider(true), &SeaweedFilerClaims{}); err != nil { + t.Fatalf("write token does not validate against the global write key: %v", err) + } + if _, err := DecodeJwt(SigningKey("side-read-key"), provider(false), &SeaweedFilerClaims{}); err != nil { + t.Fatalf("read token does not validate against the side read key: %v", err) + } +} + +func TestLoadFilerJwtFromFileEnvOverride(t *testing.T) { + configFile := filepath.Join(t.TempDir(), "security.toml") + config := ` +[jwt.filer_signing] +key = "side-write-key" +[jwt.filer_signing.read] +key = "side-read-key" +` + if err := os.WriteFile(configFile, []byte(config), 0644); err != nil { + t.Fatal(err) + } + t.Setenv("WEED_JWT_FILER_SIGNING_READ_KEY", "env-read-key") + + provider, err := LoadFilerJwtFromFile(configFile) + if err != nil { + t.Fatal(err) + } + if provider == nil { + t.Fatal("config with filer signing keys gave a nil provider") + } + + if _, err := DecodeJwt(SigningKey("env-read-key"), provider(false), &SeaweedFilerClaims{}); err != nil { + t.Fatalf("read token does not validate against the env override key: %v", err) + } + if _, err := DecodeJwt(SigningKey("side-write-key"), provider(true), &SeaweedFilerClaims{}); err != nil { + t.Fatalf("write token does not validate against the side write key: %v", err) + } +}