Merge pull request #685 from aliiohs/AddTslSupport

Ftr: add TLS support

common/constant/key.go

@@ -46,6 +46,7 @@ const (
 	DUBBO_KEY              = "dubbo"
 	RELEASE_KEY            = "release"
 	ANYHOST_KEY            = "anyhost"
+	SSL_ENABLED_KEY        = "ssl-enabled"
 )
 
 const (

config/config_loader.go

@@ -42,6 +42,7 @@ var (
 	providerConfig *ProviderConfig
 	// baseConfig = providerConfig.BaseConfig or consumerConfig
 	baseConfig *BaseConfig
+	sslEnabled = false
 
 	// configAccessMutex is used to make sure that xxxxConfig will only be created once if needed.
 	// it should be used combine with double-check to avoid the race condition
@@ -325,6 +326,12 @@ func GetBaseConfig() *BaseConfig {
 	return baseConfig
 }
 
+func GetSslEnabled() bool {
+	return sslEnabled
+}
+func SetSslEnabled(enabled bool) {
+	sslEnabled = enabled
+}
 func IsProvider() bool {
 	return providerConfig != nil
 }

config/service_config.go

@@ -188,6 +188,7 @@ func (c *ServiceConfig) Export() error {
 			common.WithPort(port),
 			common.WithParams(urlMap),
 			common.WithParamsValue(constant.BEAN_NAME_KEY, c.id),
+			common.WithParamsValue(constant.SSL_ENABLED_KEY, strconv.FormatBool(GetSslEnabled())),
 			common.WithMethods(strings.Split(methods, ",")),
 			common.WithToken(c.Token),
 		)

config/ssl_config.go

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package config
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+var (
+	serverTlsConfigBuilder getty.TlsConfigBuilder
+	clientTlsConfigBuilder getty.TlsConfigBuilder
+)
+
+func GetServerTlsConfigBuilder() getty.TlsConfigBuilder {
+	return serverTlsConfigBuilder
+}
+
+func GetClientTlsConfigBuilder() getty.TlsConfigBuilder {
+	return clientTlsConfigBuilder
+}
+
+func SetServerTlsConfigBuilder(configBuilder getty.TlsConfigBuilder) {
+	serverTlsConfigBuilder = configBuilder
+}
+
+func SetClientTlsConfigBuilder(configBuilder getty.TlsConfigBuilder) {
+	clientTlsConfigBuilder = configBuilder
+}

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/coreos/go-systemd v0.0.0-20190719114852-fd7a80b32e1f // indirect
 	github.com/creasty/defaults v1.3.0
 	github.com/docker/go-connections v0.4.0 // indirect
-	github.com/dubbogo/getty v1.3.7
+	github.com/dubbogo/getty v1.3.9
 	github.com/dubbogo/go-zookeeper v1.0.1
 	github.com/dubbogo/gost v1.9.0
 	github.com/elazarl/go-bindata-assetfs v1.0.0 // indirect
@@ -36,8 +36,6 @@ require (
 	github.com/hashicorp/vault/api v1.0.5-0.20191108163347-bdd38fca2cff // indirect
 	github.com/hashicorp/vault/sdk v0.1.14-0.20191112033314-390e96e22eb2
 	github.com/jinzhu/copier v0.0.0-20190625015134-976e0346caa8
-	github.com/juju/loggo v0.0.0-20190526231331-6e530bcce5d8 // indirect
-	github.com/juju/testing v0.0.0-20191001232224-ce9dec17d28b // indirect
 	github.com/magiconair/properties v1.8.1
 	github.com/mitchellh/hashstructure v1.0.0 // indirect
 	github.com/mitchellh/mapstructure v1.2.3
@@ -58,7 +56,6 @@ require (
 	go.uber.org/zap v1.15.0
 	google.golang.org/genproto v0.0.0-20190801165951-fa694d86fc64 // indirect
 	google.golang.org/grpc v1.23.0
-	gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce // indirect
 	gopkg.in/yaml.v2 v2.2.8
 	k8s.io/api v0.16.9
 	k8s.io/apimachinery v0.16.9

protocol/dubbo/client.go

@@ -229,6 +229,7 @@ func (c *Client) call(ct CallType, request *Request, response *Response, callbac
 	p.Service.Version = request.svcUrl.GetParam(constant.VERSION_KEY, "")
 	p.Service.Group = request.svcUrl.GetParam(constant.GROUP_KEY, "")
 	p.Service.Method = request.method
+	c.pool.sslEnabled = request.svcUrl.GetParamBool(constant.SSL_ENABLED_KEY, false)
 
 	p.Service.Timeout = c.opts.RequestTimeout
 	var timeout = request.svcUrl.GetParam(strings.Join([]string{constant.METHOD_KEYS, request.method + constant.RETRIES_KEY}, "."), "")

protocol/dubbo/pool.go

@@ -18,6 +18,7 @@
 package dubbo
 
 import (
+	"crypto/tls"
 	"fmt"
 	"math/rand"
 	"net"
@@ -33,6 +34,7 @@ import (
 
 import (
 	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/config"
 )
 
 type gettyRPCClient struct {
@@ -53,15 +55,31 @@ var (
 )
 
 func newGettyRPCClientConn(pool *gettyRPCClientPool, protocol, addr string) (*gettyRPCClient, error) {
-	c := &gettyRPCClient{
-		protocol: protocol,
-		addr:     addr,
-		pool:     pool,
-		gettyClient: getty.NewTCPClient(
+	var (
+		gettyClient getty.Client
+		sslEnabled  bool
+	)
+	sslEnabled = pool.sslEnabled
+	if sslEnabled {
+		gettyClient = getty.NewTCPClient(
+			getty.WithServerAddress(addr),
+			getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)),
+			getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval),
+			getty.WithClientSslEnabled(pool.sslEnabled),
+			getty.WithClientTlsConfigBuilder(config.GetClientTlsConfigBuilder()),
+		)
+	} else {
+		gettyClient = getty.NewTCPClient(
 			getty.WithServerAddress(addr),
 			getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)),
 			getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval),
-		),
+		)
+	}
+	c := &gettyRPCClient{
+		protocol:    protocol,
+		addr:        addr,
+		pool:        pool,
+		gettyClient: gettyClient,
 	}
 	go c.gettyClient.RunEventLoop(c.newSession)
 	idx := 1
@@ -94,16 +112,34 @@ func (c *gettyRPCClient) getActive() int64 {
 
 func (c *gettyRPCClient) newSession(session getty.Session) error {
 	var (
-		ok      bool
-		tcpConn *net.TCPConn
-		conf    ClientConfig
+		ok         bool
+		tcpConn    *net.TCPConn
+		conf       ClientConfig
+		sslEnabled bool
 	)
 
 	conf = c.pool.rpcClient.conf
+	sslEnabled = c.pool.sslEnabled
 	if conf.GettySessionParam.CompressEncoding {
 		session.SetCompressType(getty.CompressZip)
 	}
-
+	if sslEnabled {
+		if _, ok = session.Conn().(*tls.Conn); !ok {
+			panic(fmt.Sprintf("%s, session.conn{%#v} is not tls connection\n", session.Stat(), session.Conn()))
+		}
+		session.SetName(conf.GettySessionParam.SessionName)
+		session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
+		session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient))
+		session.SetEventListener(NewRpcClientHandler(c))
+		session.SetWQLen(conf.GettySessionParam.PkgWQSize)
+		session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
+		session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
+		session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
+		session.SetWaitTime(conf.GettySessionParam.waitTimeout)
+		logger.Debugf("client new session:%s\n", session.Stat())
+		session.SetTaskPool(clientGrpool)
+		return nil
+	}
 	if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
 		panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
 	}
@@ -288,9 +324,10 @@ func (c *gettyRPCClient) close() error {
 }
 
 type gettyRPCClientPool struct {
-	rpcClient *Client
-	size      int   // size of []*gettyRPCClient
-	ttl       int64 // ttl of every gettyRPCClient, it is checked when getConn
+	rpcClient  *Client
+	size       int   // size of []*gettyRPCClient
+	ttl        int64 // ttl of every gettyRPCClient, it is checked when getConn
+	sslEnabled bool
 
 	sync.Mutex
 	conns []*gettyRPCClient

protocol/dubbo/server.go

@@ -18,6 +18,7 @@
 package dubbo
 
 import (
+	"crypto/tls"
 	"fmt"
 	"net"
 )
@@ -30,6 +31,7 @@ import (
 
 import (
 	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/logger"
 	"github.com/apache/dubbo-go/config"
 )
@@ -126,7 +128,20 @@ func (s *Server) newSession(session getty.Session) error {
 	if conf.GettySessionParam.CompressEncoding {
 		session.SetCompressType(getty.CompressZip)
 	}
-
+	if _, ok = session.Conn().(*tls.Conn); ok {
+		session.SetName(conf.GettySessionParam.SessionName)
+		session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
+		session.SetPkgHandler(rpcServerPkgHandler)
+		session.SetEventListener(s.rpcHandler)
+		session.SetWQLen(conf.GettySessionParam.PkgWQSize)
+		session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
+		session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
+		session.SetCronPeriod((int)(conf.sessionTimeout.Nanoseconds() / 1e6))
+		session.SetWaitTime(conf.GettySessionParam.waitTimeout)
+		logger.Debugf("server accepts new session:%s\n", session.Stat())
+		session.SetTaskPool(srvGrpool)
+		return nil
+	}
 	if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
 		panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
 	}
@@ -163,9 +178,18 @@ func (s *Server) Start(url common.URL) {
 	)
 
 	addr = url.Location
-	tcpServer = getty.NewTCPServer(
-		getty.WithLocalAddress(addr),
-	)
+	if url.GetParamBool(constant.SSL_ENABLED_KEY, false) {
+		tcpServer = getty.NewTCPServer(
+			getty.WithLocalAddress(addr),
+			getty.WithServerSslEnabled(url.GetParamBool(constant.SSL_ENABLED_KEY, false)),
+			getty.WithServerTlsConfigBuilder(config.GetServerTlsConfigBuilder()),
+		)
+
+	} else {
+		tcpServer = getty.NewTCPServer(
+			getty.WithLocalAddress(addr),
+		)
+	}
 	tcpServer.RunEventLoop(s.newSession)
 	logger.Debugf("s bind addr{%s} ok!", addr)
 	s.tcpServer = tcpServer