diff --git a/core/discov/accountregistry.go b/core/discov/accountregistry.go index bbb1d4a9..0e102624 100644 --- a/core/discov/accountregistry.go +++ b/core/discov/accountregistry.go @@ -6,3 +6,9 @@ import "github.com/tal-tech/go-zero/core/discov/internal" func RegisterAccount(endpoints []string, user, pass string) { internal.AddAccount(endpoints, user, pass) } + +// RegisterTLS registers the CertFile/CertKeyFile/CACertFile to the given etcd. +func RegisterTLS(endpoints []string, certFile, certKeyFile, caFile string, + insecureSkipVerify bool) error { + return internal.AddTLS(endpoints, certFile, certKeyFile, caFile, insecureSkipVerify) +} diff --git a/core/discov/config.go b/core/discov/config.go index b08aea38..c1f6b7b1 100644 --- a/core/discov/config.go +++ b/core/discov/config.go @@ -4,10 +4,14 @@ import "errors" // EtcdConf is the config item with the given key on etcd. type EtcdConf struct { - Hosts []string - Key string - User string `json:",optional"` - Pass string `json:",optional"` + Hosts []string + Key string + User string `json:",optional"` + Pass string `json:",optional"` + CertFile string `json:",optional"` + CertKeyFile string `json:",optional=CertFile"` + CACertFile string `json:",optional=CertFile"` + InsecureSkipVerify bool `json:",optional"` } // HasAccount returns if account provided. @@ -15,6 +19,11 @@ func (c EtcdConf) HasAccount() bool { return len(c.User) > 0 && len(c.Pass) > 0 } +// HasTLS returns if TLS CertFile/CertKeyFile/CACertFile are provided. +func (c EtcdConf) HasTLS() bool { + return len(c.CertFile) > 0 && len(c.CertKeyFile) > 0 && len(c.CACertFile) > 0 +} + // Validate validates c. func (c EtcdConf) Validate() error { if len(c.Hosts) == 0 { diff --git a/core/discov/internal/accountmanager.go b/core/discov/internal/accountmanager.go index 4b84c9de..47b5e243 100644 --- a/core/discov/internal/accountmanager.go +++ b/core/discov/internal/accountmanager.go @@ -1,10 +1,16 @@ package internal -import "sync" +import ( + "crypto/tls" + "crypto/x509" + "io/ioutil" + "sync" +) var ( - accounts = make(map[string]Account) - lock sync.RWMutex + accounts = make(map[string]Account) + tlsConfigs = make(map[string]*tls.Config) + lock sync.RWMutex ) // Account holds the username/password for an etcd cluster. @@ -24,6 +30,32 @@ func AddAccount(endpoints []string, user, pass string) { } } +// AddTLS adds the tls cert files for the given etcd cluster. +func AddTLS(endpoints []string, certFile, certKeyFile, caFile string, insecureSkipVerify bool) error { + cert, err := tls.LoadX509KeyPair(certFile, certKeyFile) + if err != nil { + return err + } + + caData, err := ioutil.ReadFile(caFile) + if err != nil { + return err + } + + pool := x509.NewCertPool() + pool.AppendCertsFromPEM(caData) + + lock.Lock() + defer lock.Unlock() + tlsConfigs[getClusterKey(endpoints)] = &tls.Config{ + Certificates: []tls.Certificate{cert}, + RootCAs: pool, + InsecureSkipVerify: insecureSkipVerify, + } + + return nil +} + // GetAccount gets the username/password for the given etcd cluster. func GetAccount(endpoints []string) (Account, bool) { lock.RLock() @@ -32,3 +64,12 @@ func GetAccount(endpoints []string) (Account, bool) { account, ok := accounts[getClusterKey(endpoints)] return account, ok } + +// GetTLS gets the tls config for the given etcd cluster. +func GetTLS(endpoints []string) (*tls.Config, bool) { + lock.RLock() + defer lock.RUnlock() + + cfg, ok := tlsConfigs[getClusterKey(endpoints)] + return cfg, ok +} diff --git a/core/discov/internal/registry.go b/core/discov/internal/registry.go index 99d2115a..c7d6a02a 100644 --- a/core/discov/internal/registry.go +++ b/core/discov/internal/registry.go @@ -337,6 +337,9 @@ func DialClient(endpoints []string) (EtcdClient, error) { cfg.Username = account.User cfg.Password = account.Pass } + if tlsCfg, ok := GetTLS(endpoints); ok { + cfg.TLS = tlsCfg + } return clientv3.New(cfg) } diff --git a/core/discov/publisher.go b/core/discov/publisher.go index 89d5ca8e..8081e350 100644 --- a/core/discov/publisher.go +++ b/core/discov/publisher.go @@ -145,16 +145,23 @@ func (p *Publisher) revoke(cli internal.EtcdClient) { } } -// WithPubEtcdAccount provides the etcd username/password. -func WithPubEtcdAccount(user, pass string) PubOption { - return func(pub *Publisher) { - internal.AddAccount(pub.endpoints, user, pass) - } -} - // WithId customizes a Publisher with the id. func WithId(id int64) PubOption { return func(publisher *Publisher) { publisher.id = id } } + +// WithPubEtcdAccount provides the etcd username/password. +func WithPubEtcdAccount(user, pass string) PubOption { + return func(pub *Publisher) { + RegisterAccount(pub.endpoints, user, pass) + } +} + +// WithPubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile. +func WithPubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) PubOption { + return func(pub *Publisher) { + logx.Must(RegisterTLS(pub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify)) + } +} diff --git a/core/discov/subscriber.go b/core/discov/subscriber.go index 7b70bb45..5eae8449 100644 --- a/core/discov/subscriber.go +++ b/core/discov/subscriber.go @@ -5,6 +5,7 @@ import ( "sync/atomic" "github.com/tal-tech/go-zero/core/discov/internal" + "github.com/tal-tech/go-zero/core/logx" "github.com/tal-tech/go-zero/core/syncx" ) @@ -58,10 +59,17 @@ func Exclusive() SubOption { } } -// WithSubEtcdAccount customizes the Subscriber with given etcd username/password. +// WithSubEtcdAccount provides the etcd username/password. func WithSubEtcdAccount(user, pass string) SubOption { return func(sub *Subscriber) { - internal.AddAccount(sub.endpoints, user, pass) + RegisterAccount(sub.endpoints, user, pass) + } +} + +// WithSubEtcdTLS provides the etcd CertFile/CertKeyFile/CACertFile. +func WithSubEtcdTLS(certFile, certKeyFile, caFile string, insecureSkipVerify bool) SubOption { + return func(sub *Subscriber) { + logx.Must(RegisterTLS(sub.endpoints, certFile, certKeyFile, caFile, insecureSkipVerify)) } } diff --git a/zrpc/config.go b/zrpc/config.go index b47a7d2b..d076b436 100644 --- a/zrpc/config.go +++ b/zrpc/config.go @@ -83,6 +83,12 @@ func (cc RpcClientConf) BuildTarget() (string, error) { if cc.Etcd.HasAccount() { discov.RegisterAccount(cc.Etcd.Hosts, cc.Etcd.User, cc.Etcd.Pass) } + if cc.Etcd.HasTLS() { + if err := discov.RegisterTLS(cc.Etcd.Hosts, cc.Etcd.CertFile, cc.Etcd.CertKeyFile, + cc.Etcd.CACertFile, cc.Etcd.InsecureSkipVerify); err != nil { + return "", err + } + } return resolver.BuildDiscovTarget(cc.Etcd.Hosts, cc.Etcd.Key), nil } diff --git a/zrpc/internal/rpcpubserver.go b/zrpc/internal/rpcpubserver.go index f2f86735..a1773203 100644 --- a/zrpc/internal/rpcpubserver.go +++ b/zrpc/internal/rpcpubserver.go @@ -21,6 +21,10 @@ func NewRpcPubServer(etcd discov.EtcdConf, listenOn string, opts ...ServerOption if etcd.HasAccount() { pubOpts = append(pubOpts, discov.WithPubEtcdAccount(etcd.User, etcd.Pass)) } + if etcd.HasTLS() { + pubOpts = append(pubOpts, discov.WithPubEtcdTLS(etcd.CertFile, etcd.CertKeyFile, + etcd.CACertFile, etcd.InsecureSkipVerify)) + } pubClient := discov.NewPublisher(etcd.Hosts, etcd.Key, pubListenOn, pubOpts...) return pubClient.KeepAlive() }