diff --git a/weed/plugin/worker/worker.go b/weed/plugin/worker/worker.go index 279ffbe9d..5c270bfa7 100644 --- a/weed/plugin/worker/worker.go +++ b/weed/plugin/worker/worker.go @@ -14,7 +14,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "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/metadata" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -204,6 +207,12 @@ func (w *Worker) runOnce(ctx context.Context, adminAddress string) error { connCtx, cancelConn := context.WithCancel(ctx) defer cancelConn() + 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) + } + stream, err := client.WorkerStream(connCtx) if err != nil { return fmt.Errorf("open worker stream: %w", err) diff --git a/weed/worker/client.go b/weed/worker/client.go index 20fcd50aa..83aa05881 100644 --- a/weed/worker/client.go +++ b/weed/worker/client.go @@ -12,8 +12,11 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/worker/types" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" ) var ( @@ -251,6 +254,15 @@ func (c *GrpcAdminClient) handleConnect(cmd grpcCommand, s *grpcState) { cmd.resp <- nil } +// adminWorkerCredentials mints the Bearer token the admin worker gRPC service +// requires when jwt.filer_signing.key is configured; nil when it is not. +func adminWorkerCredentials() metadata.MD { + v := util.GetViper() + return security.GrpcAdminPeerCredentials( + security.SigningKey(v.GetString("jwt.filer_signing.key")), + v.GetInt("jwt.filer_signing.expires_after_seconds")) +} + // createConnection attempts to connect using the provided dial option func (c *GrpcAdminClient) createConnection() (*grpc.ClientConn, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -278,6 +290,9 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error { // Create bidirectional stream s.streamCtx, s.streamCancel = context.WithCancel(context.Background()) + if md := adminWorkerCredentials(); len(md) > 0 { + s.streamCtx = metadata.NewOutgoingContext(s.streamCtx, md) + } stream, err := s.client.WorkerStream(s.streamCtx) glog.Infof("[session %s] Worker stream created", c.sessionID) if err != nil {