plugin worker: stop forwarding the stream's admin token to job RPCs (#12391)

* plugin worker: keep the stream's admin token out of job contexts

The bearer token minted when the worker opens its control stream was
attached to connCtx, which every detection and execution context derives
from. Job RPCs therefore forwarded that connect-time token, and when a
job added its own fresh token (s3_lifecycle's LifecycleDelete) the S3
server validated the stale one first and rejected the call once it
expired.

Attach the token only to the WorkerStream call, as the Rust worker does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* credential: replace an inherited admin token instead of appending one

WithS3InternalAdminAuth appended its fresh Bearer token, so a context
that already carried an authorization value sent both, and the S3
gateway's checkAdminAuth validates only the first. Set the header on a
copy of the outgoing metadata so the fresh token is the only one sent.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* plugin worker: read the captured job metadata after the stream ends

Both result channels could be ready at once, letting select fail a
successful run. The handler reports before JobCompleted reaches admin,
so wait for the stream to finish and then read the captured metadata.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* plugin worker: test that dropping the connection cancels running jobs

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/credential/propagating_store.go

@@ -51,11 +51,12 @@ func (s *PropagatingCredentialStore) SetFilerAddressFunc(getFiler func() pb.Serv
 	}
 }
 
-// WithS3InternalAdminAuth attaches a Bearer token signed with
-// jwt.filer_signing.key to the outgoing context so the S3 gateway's internal
-// gRPC handlers (IAM cache, lifecycle) accept the call. With no key configured
-// it is a no-op, matching the S3 handler's checkAdminAuth. Returns the token's
-// lifetime (0 = no expiry) so callers can cap any downstream timeout below it.
+// WithS3InternalAdminAuth sets a fresh Bearer token signed with
+// jwt.filer_signing.key on the outgoing context, replacing any it already
+// carries, so the S3 gateway's internal gRPC handlers (IAM cache, lifecycle)
+// accept the call. With no key configured it is a no-op, matching the S3
+// handler's checkAdminAuth. Returns the token's lifetime (0 = no expiry) so
+// callers can cap any downstream timeout below it.
 func WithS3InternalAdminAuth(ctx context.Context) (context.Context, time.Duration) {
 	signingKey := util.GetViper().GetString("jwt.filer_signing.key")
 	if signingKey == "" {
@@ -66,7 +67,10 @@ func WithS3InternalAdminAuth(ctx context.Context) (context.Context, time.Duratio
 	if token == "" {
 		return ctx, 0
 	}
-	return metadata.AppendToOutgoingContext(ctx, "authorization", security.BearerPrefix+string(token)), time.Duration(expiresAfterSec) * time.Second
+	md, _ := metadata.FromOutgoingContext(ctx)
+	md = md.Copy()
+	md.Set("authorization", security.BearerPrefix+string(token))
+	return metadata.NewOutgoingContext(ctx, md), time.Duration(expiresAfterSec) * time.Second
 }
 
 func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn func(context.Context, s3_pb.SeaweedS3IamCacheClient) error) {

weed/credential/propagating_store_auth_test.go

@@ -62,3 +62,28 @@ func TestWithS3InternalAdminAuth_ZeroExpiry_ZeroTTL(t *testing.T) {
 		t.Fatalf("expected zero TTL for no-expiry token, got %v", ttl)
 	}
 }
+
+func TestWithS3InternalAdminAuth_ReplacesInheritedToken(t *testing.T) {
+	const k = "propagation-test-signing-key"
+	util.GetViper().Set("jwt.filer_signing.key", k)
+	defer util.GetViper().Set("jwt.filer_signing.key", "")
+	util.GetViper().Set("jwt.filer_signing.expires_after_seconds", 60)
+
+	parent := metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
+		"authorization", security.BearerPrefix+"stale-token",
+		"x-other", "kept",
+	))
+	ctx, _ := WithS3InternalAdminAuth(parent)
+	md, _ := metadata.FromOutgoingContext(ctx)
+	headers := md.Get("authorization")
+	if len(headers) != 1 || headers[0] == security.BearerPrefix+"stale-token" {
+		t.Fatalf("expected only the fresh token, got %v", headers)
+	}
+	if got := md.Get("x-other"); len(got) != 1 || got[0] != "kept" {
+		t.Fatalf("expected other metadata to be kept, got %v", got)
+	}
+	parentMD, _ := metadata.FromOutgoingContext(parent)
+	if got := parentMD.Get("authorization"); len(got) != 1 || got[0] != security.BearerPrefix+"stale-token" {
+		t.Fatalf("parent metadata was modified: %v", got)
+	}
+}

weed/plugin/worker/worker.go

@@ -207,13 +207,16 @@ func (w *Worker) runOnce(ctx context.Context, adminAddress string) error {
 	connCtx, cancelConn := context.WithCancel(ctx)
 	defer cancelConn()
 
+	// Only the stream carries the admin token: jobs derive their contexts from
+	// connCtx and must not forward it, since it is minted once per connection.
+	streamCtx := connCtx
 	if md := security.GrpcAdminPeerCredentials(
 		security.SigningKey(util.GetViper().GetString("jwt.filer_signing.key")),
 		util.GetViper().GetInt("jwt.filer_signing.expires_after_seconds")); len(md) > 0 {
-		connCtx = metadata.NewOutgoingContext(connCtx, md)
+		streamCtx = metadata.NewOutgoingContext(connCtx, md)
 	}
 
-	stream, err := client.WorkerStream(connCtx)
+	stream, err := client.WorkerStream(streamCtx)
 	if err != nil {
 		return fmt.Errorf("open worker stream: %w", err)
 	}

weed/plugin/worker/worker_stream_auth_test.go

@@ -0,0 +1,168 @@
+package pluginworker
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+)
+
+type authCaptureAdmin struct {
+	plugin_pb.UnimplementedPluginControlServiceServer
+	signingKey security.SigningKey
+	streamErr  chan error
+}
+
+func (s *authCaptureAdmin) WorkerStream(stream plugin_pb.PluginControlService_WorkerStreamServer) error {
+	err := s.serve(stream)
+	s.streamErr <- err
+	return err
+}
+
+func (s *authCaptureAdmin) serve(stream plugin_pb.PluginControlService_WorkerStreamServer) error {
+	if err := security.CheckGrpcAdminPeer(stream.Context(), s.signingKey); err != nil {
+		return err
+	}
+	if _, err := stream.Recv(); err != nil {
+		return err
+	}
+	if err := stream.Send(&plugin_pb.AdminToWorkerMessage{
+		RequestId: "exec-auth-1",
+		Body: &plugin_pb.AdminToWorkerMessage_ExecuteJobRequest{
+			ExecuteJobRequest: &plugin_pb.ExecuteJobRequest{
+				Job: &plugin_pb.JobSpec{JobId: "job-auth-1", JobType: "vacuum"},
+			},
+		},
+	}); err != nil {
+		return err
+	}
+	for {
+		msg, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		if msg.GetJobCompleted() != nil {
+			return nil
+		}
+	}
+}
+
+type outgoingAuthCaptureHandler struct {
+	testJobHandler
+	authorization chan []string
+}
+
+func (h *outgoingAuthCaptureHandler) Execute(ctx context.Context, _ *plugin_pb.ExecuteJobRequest, sender ExecutionSender) error {
+	md, _ := metadata.FromOutgoingContext(ctx)
+	h.authorization <- md.Get("authorization")
+	return sender.SendCompleted(&plugin_pb.JobCompleted{Success: true})
+}
+
+func startAuthCaptureWorker(t *testing.T, ctx context.Context, handler JobHandler) *authCaptureAdmin {
+	t.Helper()
+	const signingKey = "plugin-worker-stream-auth-key"
+	util.GetViper().Set("jwt.filer_signing.key", signingKey)
+	t.Cleanup(func() { util.GetViper().Set("jwt.filer_signing.key", "") })
+	util.GetViper().Set("jwt.filer_signing.expires_after_seconds", 10)
+	t.Cleanup(func() { util.GetViper().Set("jwt.filer_signing.expires_after_seconds", 0) })
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	admin := &authCaptureAdmin{signingKey: security.SigningKey(signingKey), streamErr: make(chan error, 1)}
+	server := grpc.NewServer()
+	plugin_pb.RegisterPluginControlServiceServer(server, admin)
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+
+	worker, err := NewWorker(WorkerOptions{
+		AdminServer:             listener.Addr().String(),
+		GrpcDialOption:          grpc.WithTransportCredentials(insecure.NewCredentials()),
+		Handler:                 handler,
+		MaxExecutionConcurrency: 1,
+	})
+	if err != nil {
+		t.Fatalf("NewWorker error = %v", err)
+	}
+	go worker.runOnce(ctx, listener.Addr().String())
+	return admin
+}
+
+func vacuumTestJobHandler() testJobHandler {
+	return testJobHandler{
+		capability: &plugin_pb.JobTypeCapability{JobType: "vacuum", CanExecute: true},
+		descriptor: &plugin_pb.JobTypeDescriptor{JobType: "vacuum"},
+	}
+}
+
+func TestWorkerJobContextDoesNotCarryStreamToken(t *testing.T) {
+	handler := &outgoingAuthCaptureHandler{
+		testJobHandler: vacuumTestJobHandler(),
+		authorization:  make(chan []string, 1),
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	admin := startAuthCaptureWorker(t, ctx, handler)
+
+	select {
+	case err := <-admin.streamErr:
+		if err != nil {
+			t.Fatalf("admin stream: %v", err)
+		}
+	case <-ctx.Done():
+		t.Fatal("timed out waiting for the job to complete")
+	}
+	select {
+	case got := <-handler.authorization:
+		if len(got) != 0 {
+			t.Fatalf("job context forwards the stream's admin token: %v", got)
+		}
+	default:
+		t.Fatal("job completed without running the handler")
+	}
+}
+
+type blockingCancelHandler struct {
+	testJobHandler
+	started  chan struct{}
+	canceled chan struct{}
+}
+
+func (h *blockingCancelHandler) Execute(ctx context.Context, _ *plugin_pb.ExecuteJobRequest, _ ExecutionSender) error {
+	close(h.started)
+	<-ctx.Done()
+	close(h.canceled)
+	return ctx.Err()
+}
+
+func TestWorkerConnectionTeardownCancelsJobContext(t *testing.T) {
+	handler := &blockingCancelHandler{
+		testJobHandler: vacuumTestJobHandler(),
+		started:        make(chan struct{}),
+		canceled:       make(chan struct{}),
+	}
+	connCtx, cancelConn := context.WithCancel(context.Background())
+	defer cancelConn()
+	startAuthCaptureWorker(t, connCtx, handler)
+
+	timeout := time.After(10 * time.Second)
+	select {
+	case <-handler.started:
+	case <-timeout:
+		t.Fatal("timed out waiting for the job to start")
+	}
+	cancelConn()
+	select {
+	case <-handler.canceled:
+	case <-timeout:
+		t.Fatal("job context outlived the worker connection")
+	}
+}