mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-13 01:28:37 +02:00
worker: present the admin bearer token on control streams
Workers mint a SeaweedFilerAdminClaims token when jwt.filer_signing.key is in their configuration, matching the authentication the admin server now requires.
This commit is contained in:
1 parent
55ac46f7a9
commit
613a286cde
2 files changed
+24
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user