filer.sync: sign proxied chunk I/O from the per-side security file (#11645)

* 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>

* replication: use the side filer read key for manifest downloads and fall back per access level

Manifest chunk resolution still signed proxied downloads with the
process-wide read key, so a source filer requiring its own key 401'd on
manifest-bearing files. A side security file that set only one access
level also produced empty tokens for the other instead of inheriting the
process-wide key, and the side file loader ignored the WEED_ environment
overrides the filer itself honors.

ResolveChunkManifest/ResolveOneChunkManifest keep their signatures;
FilerJwt-aware variants thread the provider down to fetchWholeChunk,
which prefers it on proxy URLs. The side loader now applies the same
environment precedence and falls back to the process-wide signer per
missing access level.

* security: verify the configured filer token lifetimes

* security: reject negative filer token lifetimes

A negative expires_after_seconds reached GenJwtForFilerServer and produced
a token with no expiration claim. Also synchronize the Authorization-header
capture in the proxy test and restore the prior viper key on cleanup.

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

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

---------

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

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

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