diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index 9e2699bd12..61fa0aa6e3 100644 --- a/weed/credential/propagating_store.go +++ b/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) { diff --git a/weed/credential/propagating_store_auth_test.go b/weed/credential/propagating_store_auth_test.go index a200d1fa6e..25aca4bb63 100644 --- a/weed/credential/propagating_store_auth_test.go +++ b/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) + } +} diff --git a/weed/plugin/worker/worker.go b/weed/plugin/worker/worker.go index 28a04c2f2c..c17bc5a531 100644 --- a/weed/plugin/worker/worker.go +++ b/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) } diff --git a/weed/plugin/worker/worker_stream_auth_test.go b/weed/plugin/worker/worker_stream_auth_test.go new file mode 100644 index 0000000000..ad8b90a712 --- /dev/null +++ b/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") + } +}