Merge pull request #323 from CodingSinger/auth

support sign and auth for request

common/constant/key.go

@@ -141,3 +141,21 @@ const (
 const (
 	TRACING_REMOTE_SPAN_CTX = "tracing.remote.span.ctx"
 )
+
+const (
+	CONSUMER_SIGN_FILTER          = "sign"
+	PROVIDER_AUTH_FILTER          = "auth"
+	SERVICE_AUTH_KEY              = "auth"
+	AUTHENTICATOR_KEY             = "authenticator"
+	DEFAULT_AUTHENTICATOR         = "accesskeys"
+	DEFAULT_ACCESS_KEY_STORAGE    = "urlstorage"
+	ACCESS_KEY_STORAGE_KEY        = "accessKey.storage"
+	REQUEST_TIMESTAMP_KEY         = "timestamp"
+	REQUEST_SIGNATURE_KEY         = "signature"
+	AK_KEY                        = "ak"
+	SIGNATURE_STRING_FORMAT       = "%s#%s#%s#%s"
+	PARAMTER_SIGNATURE_ENABLE_KEY = "param.sign"
+	CONSUMER                      = "consumer"
+	ACCESS_KEY_ID_KEY             = "accessKeyId"
+	SECRET_ACCESS_KEY_KEY         = "secretAccessKey"
+)

common/extension/auth.go

@@ -0,0 +1,32 @@
+package extension
+
+import (
+	"github.com/apache/dubbo-go/filter"
+)
+
+var (
+	authenticators    = make(map[string]func() filter.Authenticator)
+	accesskeyStorages = make(map[string]func() filter.AccessKeyStorage)
+)
+
+func SetAuthenticator(name string, fcn func() filter.Authenticator) {
+	authenticators[name] = fcn
+}
+
+func GetAuthenticator(name string) filter.Authenticator {
+	if authenticators[name] == nil {
+		panic("authenticator for " + name + " is not existing, make sure you have import the package.")
+	}
+	return authenticators[name]()
+}
+
+func SetAccesskeyStorages(name string, fcn func() filter.AccessKeyStorage) {
+	accesskeyStorages[name] = fcn
+}
+
+func GetAccesskeyStorages(name string) filter.AccessKeyStorage {
+	if accesskeyStorages[name] == nil {
+		panic("accesskeyStorages for " + name + " is not existing, make sure you have import the package.")
+	}
+	return accesskeyStorages[name]()
+}

common/url.go

@@ -339,6 +339,28 @@ func (c URL) ServiceKey() string {
 	return buf.String()
 }
 
+// ColonSeparatedKey
+// The format is "{interface}:[version]:[group]"
+func (c *URL) ColonSeparatedKey() string {
+	intf := c.GetParam(constant.INTERFACE_KEY, strings.TrimPrefix(c.Path, "/"))
+	if intf == "" {
+		return ""
+	}
+	buf := &bytes.Buffer{}
+	buf.WriteString(intf)
+	buf.WriteString(":")
+	version := c.GetParam(constant.VERSION_KEY, "")
+	if version != "" && version != "0.0.0" {
+		buf.WriteString(version)
+	}
+	group := c.GetParam(constant.GROUP_KEY, "")
+	buf.WriteString(":")
+	if group != "" {
+		buf.WriteString(group)
+	}
+	return buf.String()
+}
+
 // EncodedServiceKey ...
 func (c *URL) EncodedServiceKey() string {
 	serviceKey := c.ServiceKey()

common/url_test.go

@@ -271,3 +271,17 @@ func TestClone(t *testing.T) {
 	assert.Equal(t, u1.Protocol, "dubbo")
 	assert.Equal(t, u2.Protocol, "provider")
 }
+
+func TestColonSeparatedKey(t *testing.T) {
+	u1, _ := NewURL(context.TODO(), "dubbo://127.0.0.1:20000")
+	u1.AddParam(constant.INTERFACE_KEY, "com.ikurento.user.UserProvider")
+
+	assert.Equal(t, u1.ColonSeparatedKey(), u1.GetParam(constant.INTERFACE_KEY, "")+"::")
+	u1.AddParam(constant.VERSION_KEY, "version1")
+	assert.Equal(t, u1.ColonSeparatedKey(), u1.GetParam(constant.INTERFACE_KEY, "")+":version1:")
+	u1.AddParam(constant.GROUP_KEY, "group1")
+	assert.Equal(t, u1.ColonSeparatedKey(), u1.GetParam(constant.INTERFACE_KEY, "")+":version1:group1")
+	u1.SetParam(constant.VERSION_KEY, "")
+	assert.Equal(t, u1.ColonSeparatedKey(), u1.GetParam(constant.INTERFACE_KEY, "")+"::group1")
+
+}

config/service_config.go

@@ -67,6 +67,8 @@ type ServiceConfig struct {
 	TpsLimitRejectedHandler     string            `yaml:"tps.limit.rejected.handler" json:"tps.limit.rejected.handler,omitempty" property:"tps.limit.rejected.handler"`
 	ExecuteLimit                string            `yaml:"execute.limit" json:"execute.limit,omitempty" property:"execute.limit"`
 	ExecuteLimitRejectedHandler string            `yaml:"execute.limit.rejected.handler" json:"execute.limit.rejected.handler,omitempty" property:"execute.limit.rejected.handler"`
+	Auth                        string            `yaml:"auth" json:"auth,omitempty" property:"auth"`
+	ParamSign                   string            `yaml:"param.sign" json:"param.sign,omitempty" property:"param.sign"`
 
 	unexported    *atomic.Bool
 	exported      *atomic.Bool
@@ -220,6 +222,10 @@ func (c *ServiceConfig) getUrlMap() url.Values {
 	urlMap.Set(constant.EXECUTE_LIMIT_KEY, c.ExecuteLimit)
 	urlMap.Set(constant.EXECUTE_REJECTED_EXECUTION_HANDLER_KEY, c.ExecuteLimitRejectedHandler)
 
+	// auth filter
+	urlMap.Set(constant.SERVICE_AUTH_KEY, c.Auth)
+	urlMap.Set(constant.PARAMTER_SIGNATURE_ENABLE_KEY, c.ParamSign)
+
 	for _, v := range c.Methods {
 		prefix := "methods." + v.Name + "."
 		urlMap.Set(prefix+constant.LOADBALANCE_KEY, v.Loadbalance)

filter/access_key.go

@@ -0,0 +1,22 @@
+package filter
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+type AccessKeyPair struct {
+	AccessKey    string `yaml:"accessKey"   json:"accessKey,omitempty" property:"accessKey"`
+	SecretKey    string `yaml:"secretKey"   json:"secretKey,omitempty" property:"secretKey"`
+	ConsumerSide string `yaml:"consumerSide"   json:"ConsumerSide,consumerSide" property:"consumerSide"`
+	ProviderSide string `yaml:"providerSide"   json:"providerSide,omitempty" property:"providerSide"`
+	Creator      string `yaml:"creator"   json:"creator,omitempty" property:"creator"`
+	Options      string `yaml:"options"   json:"options,omitempty" property:"options"`
+}
+
+// AccessKeyStorage
+// This SPI Extension support us to store our AccessKeyPair or load AccessKeyPair from other
+// storage, such as filesystem.
+type AccessKeyStorage interface {
+	GetAccessKeyPair(protocol.Invocation, *common.URL) *AccessKeyPair
+}

filter/authenticator.go

@@ -0,0 +1,18 @@
+package filter
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// Authenticator
+type Authenticator interface {
+
+	// Sign
+	// give a sign to request
+	Sign(protocol.Invocation, *common.URL) error
+
+	// Authenticate
+	// verify the signature of the request is valid or not
+	Authenticate(protocol.Invocation, *common.URL) error
+}

filter/filter_impl/auth/accesskey_storage.go

@@ -0,0 +1,31 @@
+package auth
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// DefaultAccesskeyStorage
+// The default implementation of AccesskeyStorage
+type DefaultAccesskeyStorage struct {
+}
+
+// GetAccessKeyPair
+// get AccessKeyPair from url by the key "accessKeyId" and "secretAccessKey"
+func (storage *DefaultAccesskeyStorage) GetAccessKeyPair(invocation protocol.Invocation, url *common.URL) *filter.AccessKeyPair {
+	return &filter.AccessKeyPair{
+		AccessKey: url.GetParam(constant.ACCESS_KEY_ID_KEY, ""),
+		SecretKey: url.GetParam(constant.SECRET_ACCESS_KEY_KEY, ""),
+	}
+}
+
+func init() {
+	extension.SetAccesskeyStorages(constant.DEFAULT_ACCESS_KEY_STORAGE, GetDefaultAccesskeyStorage)
+}
+
+func GetDefaultAccesskeyStorage() filter.AccessKeyStorage {
+	return &DefaultAccesskeyStorage{}
+}

filter/filter_impl/auth/accesskey_storage_test.go

@@ -0,0 +1,28 @@
+package auth
+
+import (
+	"net/url"
+	"testing"
+)
+
+import (
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	invocation2 "github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestDefaultAccesskeyStorage_GetAccesskeyPair(t *testing.T) {
+	url := common.NewURLWithOptions(
+		common.WithParams(url.Values{}),
+		common.WithParamsValue(constant.SECRET_ACCESS_KEY_KEY, "skey"),
+		common.WithParamsValue(constant.ACCESS_KEY_ID_KEY, "akey"))
+	invocation := &invocation2.RPCInvocation{}
+	storage := GetDefaultAccesskeyStorage()
+	accesskeyPair := storage.GetAccessKeyPair(invocation, url)
+	assert.Equal(t, "skey", accesskeyPair.SecretKey)
+	assert.Equal(t, "akey", accesskeyPair.AccessKey)
+}

filter/filter_impl/auth/consumer_sign.go

@@ -0,0 +1,43 @@
+package auth
+
+import (
+	"context"
+	"fmt"
+)
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+// ConsumerSignFilter
+// This filter is working for signing the request on consumer side
+type ConsumerSignFilter struct {
+}
+
+func init() {
+	extension.SetFilter(constant.CONSUMER_SIGN_FILTER, getConsumerSignFilter)
+}
+
+func (csf *ConsumerSignFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	logger.Infof("invoking ConsumerSign filter.")
+	url := invoker.GetUrl()
+
+	err := doAuthWork(&url, func(authenticator filter.Authenticator) error {
+		return authenticator.Sign(invocation, &url)
+	})
+	if err != nil {
+		panic(fmt.Sprintf("Sign for invocation %s # %s failed", url.ServiceKey(), invocation.MethodName()))
+
+	}
+	return invoker.Invoke(ctx, invocation)
+}
+
+func (csf *ConsumerSignFilter) OnResponse(ctx context.Context, result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	return result
+}
+func getConsumerSignFilter() filter.Filter {
+	return &ConsumerSignFilter{}
+}

filter/filter_impl/auth/consumer_sign_test.go

@@ -0,0 +1,37 @@
+package auth
+
+import (
+	"context"
+	"testing"
+)
+
+import (
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+	"github.com/apache/dubbo-go/protocol/mock"
+)
+
+func TestConsumerSignFilter_Invoke(t *testing.T) {
+	url, _ := common.NewURL(context.TODO(), "dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider?interface=com.ikurento.user.UserProvider&group=gg&version=2.6.0")
+	url.SetParam(constant.SECRET_ACCESS_KEY_KEY, "sk")
+	url.SetParam(constant.ACCESS_KEY_ID_KEY, "ak")
+	inv := invocation.NewRPCInvocation("test", []interface{}{"OK"}, nil)
+	filter := &ConsumerSignFilter{}
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	invoker := mock.NewMockInvoker(ctrl)
+	result := &protocol.RPCResult{}
+	invoker.EXPECT().Invoke(inv).Return(result).Times(2)
+	invoker.EXPECT().GetUrl().Return(url).Times(2)
+	assert.Equal(t, result, filter.Invoke(context.Background(), invoker, inv))
+
+	url.SetParam(constant.SERVICE_AUTH_KEY, "true")
+	assert.Equal(t, result, filter.Invoke(context.Background(), invoker, inv))
+}