filer: keep the proxy-JWT test from hanging on a failed fetch

The handler sends the header non-blocking and the test stops on fetch
errors instead of waiting on a channel that may never be fed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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)

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

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
 	}

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

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.

weed/filer/filechunk_manifest_test.go

@@ -13,9 +13,12 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"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 +624,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 +658,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 +679,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 +697,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 +714,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 +733,51 @@ 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) {
+		select {
+		case gotAuth <- r.Header.Get("Authorization"):
+		default:
+		}
+		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)
+	require.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup, "5,abc", nil, false, nil, jwtFn))
+	select {
+	case auth := <-gotAuth:
+		assert.Equal(t, security.BearerPrefix+"side-read-jwt", auth)
+	default:
+		t.Fatal("the proxied fetch never reached the handler")
+	}
+
+	// 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 ""
+	}))
+}

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())

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)

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,

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
 	}

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

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

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