Repository: lf-edge/ekuiper
Pull request #1042 (2021-10-20 03:01:10)

source pool detach problem and test fix


Patch:
diff --git a/docs/en_US/edgex/edgex_meta.md b/docs/en_US/edgex/edgex_meta.md
index 4ed5e269..a680f2f1 100644
--- a/docs/en_US/edgex/edgex_meta.md
+++ b/docs/en_US/edgex/edgex_meta.md
@@ -43,7 +43,7 @@ So how the EdgeX data are managed in eKuiper? Let's take an example.
 
 As in below - firstly, user creates an EdgeX stream named ``events`` with yellow color.
 
-<img src="create_stream.png" style="zoom:50%;" />
+<img src="./create_stream.png" style="zoom:50%;" />
 
 Secondly, one message is published to message bus as in below. 
 
@@ -51,7 +51,7 @@ Secondly, one message is published to message bus as in below.
 - Reading name ``temperature`` & ``Humidity`` with red color. 
 - It has some ``metadata`` that is not necessary to "visible", but it probably will be used during data analysis, such as ``DeviceName`` field in ``Event`` structure. eKuiper saves these values into message tuple named metadata, and user can get these values during analysis. **Notice that, metadata name `DeviceName` was renamed from `Device` in EdgeX v2.**
 
-<img src="bus_data.png" style="zoom:50%;" />
+<img src="./bus_data.png" style="zoom:50%;" />
 
 Thirdly, a SQL is provided for data analysis. Please notice that,
 
@@ -59,7 +59,7 @@ Thirdly, a SQL is provided for data analysis. Please notice that,
 - The SELECT fields ``temperature`` & ``humidity`` are red color, which are the ``Name`` field of readings.
 - The WHERE clause ``meta(deviceName)`` in green color, which is ued for extracting ``DeviceName`` field from ``Events`` structure. The SQL statement will filter data that device names are not ``demo``.
 
-<img src="sql.png" style="zoom:50%;" />
+<img src="./sql.png" style="zoom:50%;" />
 
 Below are some other samples that extract other metadata through ``meta`` function.
 
diff --git a/docs/en_US/operation/configuration_file.md b/docs/en_US/operation/configuration_file.md
index 608293df..d750d0ef 100644
--- a/docs/en_US/operation/configuration_file.md
+++ b/docs/en_US/operation/configuration_file.md
@@ -65,6 +65,14 @@ The port for the rest api http server to listen to.
 ### restTls
 The tls cert file path and key file path setting. If restTls is not set, the rest api server will listen on http. Otherwise, it will listen on https.
 
+## authentication 
+eKuiper will check the ``Token`` for rest api when ``authentication`` option is true. please check this file for [more info](./operations.md).
+
+```yaml
+basic:
+  authentication: false
+```
+
 ## Prometheus Configuration
 
 eKuiper can export metrics to prometheus if ``prometheus`` option is true. The prometheus will be served with the port specified by ``prometheusPort`` option.
diff --git a/docs/en_US/operation/operations.md b/docs/en_US/operation/operations.md
index 564de3ec..56d8c583 100644
--- a/docs/en_US/operation/operations.md
+++ b/docs/en_US/operation/operations.md
@@ -7,3 +7,49 @@
 
 eKuiper provides some RESTful management APIs. Please refer to [Rest-API doc](../restapi/overview.md) for more detailed information.
 
+
+## Authentication
+
+eKuiper support ``JWT RSA256`` authentication for the RESTful management APIs since ``1.4.0`` if enabled . Users need put their Public Key in ``etc/mgmt`` folder and use the corresponding Private key to sign the JWT Tokens.
+When user request the RESTful apis, put the ``Token`` in http request headers in the following format:
+```go
+Authorization: XXXXXXXXXXXXXXX
+```
+If the token is correct, eKuiper will respond the result; otherwise, it will return http ``401``code.
+
+
+### JWT Header
+
+```json
+{
+  "typ": "JWT",
+  "alg": "RS256"
+}
+```
+
+
+### JWT payload
+The JWT Payload should use the following format
+
+|  field   | optional |  meaning  |
+|  ----  | ----  | ----  |
+| iss  | false| Issuer , must use the same name with the public key put in ``etc/mgmt``|
+| aud  | false |Audience , must be ``eKuiper`` |
+| exp  | true |Expiration Time |
+| jti  | true |JWT ID |
+| iat  | true |Issued At |
+| nbf  | true |Not Before |
+| sub  | true |Subject |
+
+There is an example in json format
+```json
+{
+  "iss": "sample_key.pub",
+  "adu": "eKuiper"
+}
+```
+When use this format, user must make sure the correct Public key file ``sample_key.pub`` are under ``etc/mgmt`` .
+
+### JWT Signature
+
+need use the Private key to sign the Tokens and put the corresponding Public Key in ``etc/mgmt`` .
\ No newline at end of file
diff --git a/docs/zh_CN/edgex/edgex_meta.md b/docs/zh_CN/edgex/edgex_meta.md
index e83e091a..68590048 100644
--- a/docs/zh_CN/edgex/edgex_meta.md
+++ b/docs/zh_CN/edgex/edgex_meta.md
@@ -43,7 +43,7 @@
 
 如下所示，首先用户创建了一个名为 ``events`` 的 EdgeX 流定义（以黄色高亮标示）。
 
-<img src="create_stream.png" style="zoom:50%;" />
+<img src="./create_stream.png" style="zoom:50%;" />
 
 其次，如下所示，一条消息被发送到消息总线。
 
@@ -51,7 +51,7 @@
 - Reading 名称为 ``temperature`` & ``Humidity`` ，用红色高亮标示
 - 这里有些 ``元数据`` 是没有必要「可见」的，但是这些值在分析的时候可能会被用到，例如``Event`` 结构体中的 ``DeviceName`` 字段。eKuiper 将这些值保存在 eKuiper 消息中的名为 metadata 的字段中，用户在分析阶段可以获取到这些值。**需要注意的是, EdgeX v2 中元数据 `Device` 已重命名为 `DeviceName` 。**
 
-<img src="bus_data.png" style="zoom:50%;" />
+<img src="./bus_data.png" style="zoom:50%;" />
 
 最后，提供一条 SQL 用于数据分析，此处请注意，
 
@@ -59,7 +59,7 @@
 - SELECT 中的 ``temperature`` & ``humidity`` 字段为红色高亮，它们是 readings 中的 ``Name`` 字段的值。
 - WHERE 子句中的 ``meta(deviceName)`` 为绿色高亮，用于从 ``Events ``结构体中抽取 ``device`` 字段。该 SQL 语句将过滤所有设备名称不是 ``demo`` 的记录。
 
-<img src="sql.png" style="zoom:50%;" />
+<img src="./sql.png" style="zoom:50%;" />
 
 以下是使用 ``meta`` 函数抽取别的元数据的一些例子。
 
@@ -91,4 +91,3 @@ Kuper 的 ``meta`` 函数可以用于访问元数据，以下列出了所有在
 
 - Events: id, deviceName, profileName, sourceName, origin, tags, correlationid
 - Readning: id, deviceName, profileName, origin, valueType
-
diff --git a/docs/zh_CN/operation/configuration_file.md b/docs/zh_CN/operation/configuration_file.md
index 626912e9..329672f1 100755
--- a/docs/zh_CN/operation/configuration_file.md
+++ b/docs/zh_CN/operation/configuration_file.md
@@ -45,6 +45,15 @@ REST http 服务器监听端口
 ### restTls
 TLS 证书 cert 文件和 key 文件位置。如果 restTls 选项未配置，则 REST 服务器将启动为 http 服务器，否则启动为 https 服务器。
 
+## authentication 
+当 ``authentication`` 选项为 true 时，eKuiper 将为 rest api 请求检查 ``Token`` 。请检查此文件以获取 [更多信息](./operations.md)。
+
+```yaml
+basic:
+  authentication: false
+```
+
+
 ## Prometheus 配置
 
 如果 `prometheus` 参数设置为 true，eKuiper 将把运行指标暴露到 prometheus。Prometheus 将运行在 `prometheusPort` 参数指定的端口上。
diff --git a/docs/zh_CN/operation/operations.md b/docs/zh_CN/operation/operations.md
index db95a90e..01d93778 100755
--- a/docs/zh_CN/operation/operations.md
+++ b/docs/zh_CN/operation/operations.md
@@ -5,4 +5,50 @@
 
 ## Restful APIs
 
-eKuiper 提供了一些 RESTful 管理 API。eKuiper 提供了一些 RESTful 管理 APIs。请参考 [Rest-API 文档](../restapi/overview.md)以获取更详细信息。
\ No newline at end of file
+eKuiper 提供了一些 RESTful 管理 API。eKuiper 提供了一些 RESTful 管理 APIs。请参考 [Rest-API 文档](../restapi/overview.md)以获取更详细信息。
+
+## Authentication
+
+如果使能的话， eKuiper 从 1.4.0 起将为 RESTful API 提供基于 ``JWT RSA256`` 的身份验证。用户需要将他们的公钥放在 ``etc/mgmt`` 文件夹中，并使用相应的私钥来签署 JWT 令牌。
+当用户请求 RESTful API 时，将 ``Token`` 放在 http 请求头中，格式如下：
+```
+Authorization：XXXXXXXXXXXXXXX
+```
+如果token正确，eKuiper会响应结果；否则，它将返回 http ``401`` 代码。
+
+
+### JWT Header
+
+```json
+{
+  "typ": "JWT",
+  "alg": "RS256"
+}
+```
+
+
+### JWT Payload
+JWT Payload 应使用以下格式
+
+|  字段   | 是否可选 |  意义  |
+|  ----  | ----  | ----  |
+| iss  | 否| 颁发者 ,  此字段必须与``etc/mgmt``目录中的相应公钥文件名字一致|
+| aud  | 否 |颁发对象 , 此字段必须是 ``eKuiper`` |
+| exp  | 是 |过期时间 |
+| jti  | 是 |JWT ID |
+| iat  | 是 |颁发时间 |
+| nbf  | 是 |Not Before |
+| sub  | 是 |主题 |
+
+这里有一个 json 格式的例子
+```json
+{
+  "iss": "sample_key.pub",
+  "adu": "eKuiper"
+}
+```
+使用此格式时，用户必须确保正确的公钥文件 ``sample_key.pub`` 位于 ``etc/mgmt`` 下。
+
+### JWT Signature
+
+需要使用私钥对令牌进行签名，并将相应的公钥放在 ``etc/mgmt`` 中。
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index ce2b08ea..ccfc7c83 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -17,6 +17,8 @@ basic:
   restIp: 0.0.0.0
   # REST service port
   restPort: 9081
+  # true|false, when true, will check the RSA jwt token for rest api
+  authentication: false
   #  restTls:
   #    certfile: /var/https-server.crt
   #    keyfile: /var/https-server.key
diff --git a/etc/mgmt/sample_key b/etc/mgmt/sample_key
new file mode 100644
index 00000000..e69284a3
--- /dev/null
+++ b/etc/mgmt/sample_key
@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEA4f5wg5l2hKsTeNem/V41fGnJm6gOdrj8ym3rFkEU/wT8RDtn
+SgFEZOQpHEgQ7JL38xUfU0Y3g6aYw9QT0hJ7mCpz9Er5qLaMXJwZxzHzAahlfA0i
+cqabvJOMvQtzD6uQv6wPEyZtDTWiQi9AXwBpHssPnpYGIn20ZZuNlX2BrClciHhC
+PUIIZOQn/MmqTD31jSyjoQoV7MhhMTATKJx2XrHhR+1DcKJzQBSTAGnpYVaqpsAR
+ap+nwRipr3nUTuxyGohBTSmjJ2usSeQXHI3bODIRe1AuTyHceAbewn8b462yEWKA
+Rdpd9AjQW5SIVPfdsz5B6GlYQ5LdYKtznTuy7wIDAQABAoIBAQCwia1k7+2oZ2d3
+n6agCAbqIE1QXfCmh41ZqJHbOY3oRQG3X1wpcGH4Gk+O+zDVTV2JszdcOt7E5dAy
+MaomETAhRxB7hlIOnEN7WKm+dGNrKRvV0wDU5ReFMRHg31/Lnu8c+5BvGjZX+ky9
+POIhFFYJqwCRlopGSUIxmVj5rSgtzk3iWOQXr+ah1bjEXvlxDOWkHN6YfpV5ThdE
+KdBIPGEVqa63r9n2h+qazKrtiRqJqGnOrHzOECYbRFYhexsNFz7YT02xdfSHn7gM
+IvabDDP/Qp0PjE1jdouiMaFHYnLBbgvlnZW9yuVf/rpXTUq/njxIXMmvmEyyvSDn
+FcFikB8pAoGBAPF77hK4m3/rdGT7X8a/gwvZ2R121aBcdPwEaUhvj/36dx596zvY
+mEOjrWfZhF083/nYWE2kVquj2wjs+otCLfifEEgXcVPTnEOPO9Zg3uNSL0nNQghj
+FuD3iGLTUBCtM66oTe0jLSslHe8gLGEQqyMzHOzYxNqibxcOZIe8Qt0NAoGBAO+U
+I5+XWjWEgDmvyC3TrOSf/KCGjtu0TSv30ipv27bDLMrpvPmD/5lpptTFwcxvVhCs
+2b+chCjlghFSWFbBULBrfci2FtliClOVMYrlNBdUSJhf3aYSG2Doe6Bgt1n2CpNn
+/iu37Y3NfemZBJA7hNl4dYe+f+uzM87cdQ214+jrAoGAXA0XxX8ll2+ToOLJsaNT
+OvNB9h9Uc5qK5X5w+7G7O998BN2PC/MWp8H+2fVqpXgNENpNXttkRm1hk1dych86
+EunfdPuqsX+as44oCyJGFHVBnWpm33eWQw9YqANRI+pCJzP08I5WK3osnPiwshd+
+hR54yjgfYhBFNI7B95PmEQkCgYBzFSz7h1+s34Ycr8SvxsOBWxymG5zaCsUbPsL0
+4aCgLScCHb9J+E86aVbbVFdglYa5Id7DPTL61ixhl7WZjujspeXZGSbmq0Kcnckb
+mDgqkLECiOJW2NHP/j0McAkDLL4tysF8TLDO8gvuvzNC+WQ6drO2ThrypLVZQ+ry
+eBIPmwKBgEZxhqa0gVvHQG/7Od69KWj4eJP28kq13RhKay8JOoN0vPmspXJo1HY3
+CKuHRG+AP579dncdUnOMvfXOtkdM4vk0+hWASBQzM9xzVcztCa+koAugjVaLS9A+
+9uQoqEeVNTckxx0S2bYevRy7hGQmUJTyQm3j1zEUR5jpdbL83Fbq
+-----END RSA PRIVATE KEY-----
\ No newline at end of file
diff --git a/etc/mgmt/sample_key.pub b/etc/mgmt/sample_key.pub
new file mode 100644
index 00000000..d5b7f710
--- /dev/null
+++ b/etc/mgmt/sample_key.pub
@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4f5wg5l2hKsTeNem/V41
+fGnJm6gOdrj8ym3rFkEU/wT8RDtnSgFEZOQpHEgQ7JL38xUfU0Y3g6aYw9QT0hJ7
+mCpz9Er5qLaMXJwZxzHzAahlfA0icqabvJOMvQtzD6uQv6wPEyZtDTWiQi9AXwBp
+HssPnpYGIn20ZZuNlX2BrClciHhCPUIIZOQn/MmqTD31jSyjoQoV7MhhMTATKJx2
+XrHhR+1DcKJzQBSTAGnpYVaqpsARap+nwRipr3nUTuxyGohBTSmjJ2usSeQXHI3b
+ODIRe1AuTyHceAbewn8b462yEWKARdpd9AjQW5SIVPfdsz5B6GlYQ5LdYKtznTuy
+7wIDAQAB
+-----END PUBLIC KEY-----
\ No newline at end of file
diff --git a/go.mod b/go.mod
index 12fb3e10..03b7ede2 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/alicebob/miniredis/v2 v2.15.1
 	github.com/benbjohnson/clock v1.0.0
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0
 	github.com/edgexfoundry/go-mod-messaging/v2 v2.0.1
@@ -32,7 +33,6 @@ require (
 	github.com/msgpack-rpc/msgpack-rpc-go v0.0.0-20131026060856-c76397e1782b
 	github.com/msgpack/msgpack-go v0.0.0-20130625150338-8224460e6fa3 // indirect
 	github.com/pebbe/zmq4 v1.2.7
-	github.com/pkg/errors v0.8.1 // indirect
 	github.com/prometheus/client_golang v1.2.1
 	github.com/sirupsen/logrus v1.4.2
 	github.com/smartystreets/goconvey v1.6.4 // indirect
diff --git a/go.sum b/go.sum
index 3c260a3c..efce374d 100644
--- a/go.sum
+++ b/go.sum
@@ -43,6 +43,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
+github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/eclipse/paho.mqtt.golang v1.3.5 h1:sWtmgNxYM9P2sP+xEItMozsR3w0cqZFlqnNN1bdl41Y=
 github.com/eclipse/paho.mqtt.golang v1.3.5/go.mod h1:eTzb4gxwwyWpqBUHGQZ4ABAV7+Jgm1PklsYT/eo8Hcc=
 github.com/edgexfoundry/go-mod-core-contracts/v2 v2.0.0 h1:tvfovdyoHOb392L59hiuA90awiXLX5IR3HOgbcWZkVQ=
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 481a22ea..e8a39179 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -52,6 +52,7 @@ type KuiperConf struct {
 		Prometheus     bool     `yaml:"prometheus"`
 		PrometheusPort int      `yaml:"prometheusPort"`
 		PluginHosts    string   `yaml:"pluginHosts"`
+		Authentication bool     `yaml:"authentication"`
 	}
 	Rule api.RuleOption
 	Sink struct {
diff --git a/internal/pkg/jwt/jwt_rsa.go b/internal/pkg/jwt/jwt_rsa.go
new file mode 100644
index 00000000..d04433e7
--- /dev/null
+++ b/internal/pkg/jwt/jwt_rsa.go
@@ -0,0 +1,85 @@
+package jwt
+
+import (
+	"errors"
+	"fmt"
+	"github.com/dgrijalva/jwt-go"
+	"time"
+)
+
+const ExpireTimeMinutes = 10
+
+type Token struct {
+	jwt.StandardClaims
+}
+
+type ErrorType int8
+
+const (
+	JWT_VALIDATE_ERROR ErrorType = 1
+)
+
+const JWT_VALIDATE_TEMP = "JWTVAL__ERRCODE:%d__ERRSTR:%s__TOKEN:%s"
+const JWT_OTHER_TEMP = "JWTOTH__ERRSTR:%s__TOKEN:%s"
+
+type Error struct {
+	errType   ErrorType
+	Inner     error
+	metaToken string
+}
+
+func (e Error) Error() string {
+	switch e.errType {
+	case JWT_VALIDATE_ERROR:
+		if ve, ok := e.Inner.(*jwt.ValidationError); ok {
+			return fmt.Sprintf(JWT_VALIDATE_TEMP, ve.Errors, ve.Error(), e.metaToken)
+		}
+	default:
+		return fmt.Sprintf(JWT_OTHER_TEMP, e.Inner.Error(), e.metaToken)
+	}
+	return "Invalid token " + e.metaToken
+}
+
+func CreateToken(signKeyName, issuer, aud string) (string, error) {
+	tk := &Token{}
+	tk.Issuer = issuer
+	tk.Audience = aud
+	tk.ExpiresAt = time.Now().Add(time.Duration(ExpireTimeMinutes) * time.Minute).Unix()
+	token := jwt.NewWithClaims(jwt.GetSigningMethod("RS256"), tk)
+	signKey, err := GetPrivateKeyWithKeyName(signKeyName)
+	if err != nil {
+		return "", err
+	}
+	return token.SignedString(signKey)
+}
+
+func ParseToken(th string) (*Token, error) {
+	tk := &Token{}
+	token, err := jwt.ParseWithClaims(th, tk, func(token *jwt.Token) (interface{}, error) {
+		jwtToken := token.Claims.(*Token)
+
+		if jwtToken.Issuer == "" {
+			return "", fmt.Errorf("issuer field not exist in jwt payload")
+		}
+		pubKey, err := GetPublicKey(jwtToken.Issuer)
+		if err != nil {
+			return "", err
+		}
+		return pubKey, nil
+	})
+	if ve, ok := err.(*jwt.ValidationError); ok {
+		return tk, Error{
+			errType:   JWT_VALIDATE_ERROR,
+			Inner:     ve,
+			metaToken: th,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+	if !token.Valid {
+		return nil, errors.New("invalid token")
+	}
+	return tk, nil
+}
diff --git a/internal/pkg/jwt/rsa_keys.go b/internal/pkg/jwt/rsa_keys.go
new file mode 100644
index 00000000..32892fa9
--- /dev/null
+++ b/internal/pkg/jwt/rsa_keys.go
@@ -0,0 +1,99 @@
+package jwt
+
+import (
+	"crypto/rsa"
+	"fmt"
+	"github.com/dgrijalva/jwt-go"
+	"github.com/lf-edge/ekuiper/internal/conf"
+	"io/ioutil"
+	"path"
+	"strings"
+	"sync"
+)
+
+var privateKeyRepository = make(map[string]*rsa.PrivateKey)
+var repositoryLock sync.Mutex
+
+const RSAKeyDir = "mgmt"
+
+func GetPrivateKeyWithKeyName(keyName string) (*rsa.PrivateKey, error) {
+	repositoryLock.Lock()
+	defer repositoryLock.Unlock()
+
+	key, ok := privateKeyRepository[keyName]
+	if ok {
+		return key, nil
+	}
+
+	privateKey, err := privateKeyFromFile(keyName)
+	if err != nil {
+		return nil, err
+	}
+
+	privateKeyRepository[keyName] = privateKey
+
+	return privateKey, nil
+}
+
+func GetPublicKey(keyName string) (*rsa.PublicKey, error) {
+	publicKey, err := publicKeyFromFile(keyName)
+	if err != nil {
+		return nil, err
+	}
+
+	return publicKey, nil
+}
+
+func insensitiveGetFilePath(prikeyName string) (string, error) {
+	confDir, err := conf.GetConfLoc()
+	if nil != err {
+		return "", err
+	}
+
+	dir := path.Join(confDir, RSAKeyDir)
+	infos, err := ioutil.ReadDir(dir)
+	if nil != err {
+		return "", err
+	}
+
+	for _, info := range infos {
+		fileName := info.Name()
+		if strings.ToLower(fileName) == strings.ToLower(prikeyName) {
+			filePath := path.Join(dir, fileName)
+			return filePath, nil
+		}
+	}
+	return "", fmt.Errorf("not found target key file %s in /etc/%s", prikeyName, RSAKeyDir)
+}
+
+func privateKeyFromFile(keyName string) (*rsa.PrivateKey, error) {
+	keyPath, err := insensitiveGetFilePath(keyName)
+	if err != nil {
+		return nil, err
+	}
+	keyBytes, err := ioutil.ReadFile(keyPath)
+	if err != nil {
+		return nil, err
+	}
+	signKey, err := jwt.ParseRSAPrivateKeyFromPEM(keyBytes)
+	if err != nil {
+		return nil, err
+	}
+	return signKey, nil
+}
+
+func publicKeyFromFile(keyName string) (*rsa.PublicKey, error) {
+	keyPath, err := insensitiveGetFilePath(keyName)
+	if err != nil {
+		return nil, err
+	}
+	keyBytes, err := ioutil.ReadFile(keyPath)
+	if err != nil {
+		return nil, err
+	}
+	pubKey, err := jwt.ParseRSAPublicKeyFromPEM(keyBytes)
+	if err != nil {
+		return nil, err
+	}
+	return pubKey, nil
+}
diff --git a/internal/server/middleware/auth.go b/internal/server/middleware/auth.go
new file mode 100644
index 00000000..0204e716
--- /dev/null
+++ b/internal/server/middleware/auth.go
@@ -0,0 +1,39 @@
+package middleware
+
+import (
+	"fmt"
+	"github.com/lf-edge/ekuiper/internal/pkg/jwt"
+	"net/http"
+)
+
+var notAuth = []string{"/", "/ping"}
+
+var Auth = func(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestPath := r.URL.Path
+		for _, value := range notAuth {
+			if value == requestPath {
+				next.ServeHTTP(w, r)
+				return
+			}
+		}
+
+		tokenHeader := r.Header.Get("Authorization")
+
+		if tokenHeader == "" {
+			http.Error(w, "missing_token", http.StatusUnauthorized)
+			return
+		}
+		tk, err := jwt.ParseToken(tokenHeader)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnauthorized)
+			return
+		}
+		if tk.StandardClaims.Audience != "eKuiper" {
+			http.Error(w, fmt.Sprintf("audience field should be eKuiper, but got %s", tk.StandardClaims.Audience), http.StatusUnauthorized)
+			return
+		}
+
+		next.ServeHTTP(w, r)
+	})
+}
diff --git a/internal/server/rest.go b/internal/server/rest.go
index 968f9682..21e0520e 100644
--- a/internal/server/rest.go
+++ b/internal/server/rest.go
@@ -25,6 +25,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/plugin"
 	"github.com/lf-edge/ekuiper/internal/plugin/native"
 	"github.com/lf-edge/ekuiper/internal/plugin/portable"
+	"github.com/lf-edge/ekuiper/internal/server/middleware"
 	"github.com/lf-edge/ekuiper/internal/service"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
@@ -91,7 +92,7 @@ func jsonResponse(i interface{}, w http.ResponseWriter, logger api.Logger) {
 	}
 }
 
-func createRestServer(ip string, port int) *http.Server {
+func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r := mux.NewRouter()
 	r.HandleFunc("/", rootHandler).Methods(http.MethodGet, http.MethodPost)
 	r.HandleFunc("/ping", pingHandler).Methods(http.MethodGet)
@@ -138,6 +139,10 @@ func createRestServer(ip string, port int) *http.Server {
 	r.HandleFunc("/services/functions/{name}", serviceFunctionHandler).Methods(http.MethodGet)
 	r.HandleFunc("/services/{name}", serviceHandler).Methods(http.MethodDelete, http.MethodGet, http.MethodPut)
 
+	if needToken {
+		r.Use(middleware.Auth)
+	}
+
 	server := &http.Server{
 		Addr: fmt.Sprintf("%s:%d", ip, port),
 		// Good practice to set timeouts to avoid Slowloris attacks.
diff --git a/internal/server/server.go b/internal/server/server.go
index 86cde170..c8c0360a 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -133,7 +133,7 @@ func StartUp(Version, LoadFileType string) {
 	}
 
 	//Start rest service
-	srvRest := createRestServer(conf.Config.Basic.RestIp, conf.Config.Basic.RestPort)
+	srvRest := createRestServer(conf.Config.Basic.RestIp, conf.Config.Basic.RestPort, conf.Config.Basic.Authentication)
 	go func() {
 		var err error
 		if conf.Config.Basic.RestTls == nil {
diff --git a/internal/topo/node/source_pool.go b/internal/topo/node/source_pool.go
index e7bd5abd..1c5729c8 100644
--- a/internal/topo/node/source_pool.go
+++ b/internal/topo/node/source_pool.go
@@ -205,22 +205,19 @@ func (ss *sourceSingleton) run(name, key string) {
 
 func (ss *sourceSingleton) broadcast(val api.SourceTuple) {
 	logger := ss.ctx.GetLogger()
-	var wg sync.WaitGroup
 	ss.RLock()
-	wg.Add(len(ss.outputs))
 	for n, out := range ss.outputs {
-		go func(name string, output chan<- api.SourceTuple) {
+		go func(name string, dataCh *DynamicChannelBuffer) {
 			select {
-			case output <- val:
+			case dataCh.Out <- val:
 				logger.Debugf("broadcast from source pool to %s done", name)
 			case <-ss.ctx.Done():
-				// rule stop so stop waiting
+			case <-dataCh.done:
+				// detached
 			}
-			wg.Done()
-		}(n, out.dataCh.Out)
+		}(n, out.dataCh)
 	}
 	ss.RUnlock()
-	wg.Wait()
 }
 
 func (ss *sourceSingleton) broadcastError(err error) {


Test patch:
diff --git a/internal/pkg/jwt/jwt_rsa_test.go b/internal/pkg/jwt/jwt_rsa_test.go
new file mode 100644
index 00000000..c43a6806
--- /dev/null
+++ b/internal/pkg/jwt/jwt_rsa_test.go
@@ -0,0 +1,76 @@
+package jwt
+
+import (
+	"fmt"
+	"testing"
+)
+
+var expiredToken = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJhdWQiOiJlS3VpcGVyIiwiZXhwIjoxNjM2MDExMzQxLCJpc3MiOiJzYW1wbGVfa2V5LnB1YiJ9.qm5Pq9VxDC10qbOM081U5NwScTOxYV_F5vyqbU9rXB2ebz4kDio_R2tgEgGyJ41lwD7gFl1quBjp_EgokPZNOoGRg5R1Ygf7iF8XJSDxYkspSCsBtZAuMCo3MCz3slQyvnr24qv3idUDhlwO6FPHGLaLHEyvrETSl1ZcECq2wvW01Tc2Jmg0-Kpp6TmEbH5aD-L0or5Bfy0ytBQ64nd2hKVaoADZZOXSt1iH2-1R35fEc_lBw7zs4QpCC2R--muoqYsYkESR08o6wIKAxRJvqeWab3C9k_g0zaPhwa7ZQ9wRzah-tc6PdotZkAyH7BCx-f7llO7UT47k0GnrhBe21g"
+var badFormatToken = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJmb28iOiJiYXIiLCJleHAiOjE1MDAwLCJpc3MiOiJ0ZXN0In0"
+
+func genToken(signKeyName, issuer, aud string) string {
+	tkStr, _ := CreateToken(signKeyName, issuer, aud)
+	return tkStr
+}
+
+func TestParseToken(t *testing.T) {
+	type args struct {
+		th string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "pass: have issuer public key",
+			args: args{
+				th: genToken("sample_key", "sample_key.pub", "eKuiper"),
+			},
+			wantErr: false,
+		},
+		{
+			name: "fail: token expired",
+			args: args{
+				th: expiredToken,
+			},
+			wantErr: true,
+		},
+		{
+			name: "fail: token sign error",
+			args: args{
+				th: genToken("sample_key", "sample_key.pub", "eKuiper") + "badSign",
+			},
+			wantErr: true,
+		},
+		{
+			name: "fail: do not have issuer's public key",
+			args: args{
+				th: genToken("sample_key", "notexist.pub", "eKuiper"),
+			},
+			wantErr: true,
+		},
+		{
+			name: "bad token format",
+			args: args{
+				th: badFormatToken,
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := ParseToken(tt.args.th)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ParseToken() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			if err != nil {
+				fmt.Printf("=====================\n")
+				fmt.Printf("Validate Error %s", err)
+			}
+
+		})
+	}
+}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager_test.go b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
index ba872b69..dfd2de0e 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager_test.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
@@ -107,6 +107,7 @@ func TestPluginInstance(t *testing.T) {
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		err := ins.StartSymbol(sctx, tests[0].c)
 		if err != nil {
 			t.Errorf("start command err %v", err)
@@ -120,13 +121,12 @@ func TestPluginInstance(t *testing.T) {
 			}
 			err = ins.StopSymbol(sctx, tt.c)
 			if err != nil {
-				t.Errorf("stop command err %v", err)
-				return
+				fmt.Printf("stop command err %v\n", err)
+				continue
 			}
 		}
-		wg.Done()
 	}()
-	// start symbol1 to avoild instance clean
+	// start symbol1 to avoid instance clean
 	msg, err := client.Recv()
 	if err != nil {
 		t.Errorf("receive start command err %v", err)
diff --git a/internal/plugin/portable/test/portable_rule_test.go b/internal/plugin/portable/test/portable_rule_test.go
index 15cf8a20..a20f9549 100644
--- a/internal/plugin/portable/test/portable_rule_test.go
+++ b/internal/plugin/portable/test/portable_rule_test.go
@@ -142,7 +142,7 @@ func TestSourceAndFunc(t *testing.T) {
 			if compareMetrics(tp, tt.M) {
 				cancel()
 				// need to wait for file results
-				time.Sleep(10 * time.Millisecond)
+				time.Sleep(100 * time.Millisecond)
 				results := getResults(fmt.Sprintf("cache%d", i+1))
 				fmt.Printf("get results %v\n", results)
 				time.Sleep(10 * time.Millisecond)
@@ -160,7 +160,7 @@ func TestSourceAndFunc(t *testing.T) {
 					mm = append(mm, mapRes)
 				}
 				if !reflect.DeepEqual(tt.R, mm) {
-					t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.Rule, tt.R, results)
+					t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.Rule, tt.R, mm)
 				}
 				break
 			}
diff --git a/internal/server/middleware/auth_test.go b/internal/server/middleware/auth_test.go
new file mode 100644
index 00000000..7f0e2d4f
--- /dev/null
+++ b/internal/server/middleware/auth_test.go
@@ -0,0 +1,84 @@
+package middleware
+
+import (
+	"github.com/lf-edge/ekuiper/internal/pkg/jwt"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+)
+
+func genToken(signKeyName, issuer, aud string) string {
+	tkStr, _ := jwt.CreateToken(signKeyName, issuer, aud)
+	return tkStr
+}
+
+func Test_AUTH(t *testing.T) {
+	nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+	})
+
+	handler := Auth(nextHandler)
+
+	type args struct {
+		th string
+	}
+	tests := []struct {
+		name     string
+		args     args
+		req      *http.Request
+		res      *httptest.ResponseRecorder
+		wantCode int
+	}{
+		{
+			name:     "token right",
+			args:     args{th: genToken("sample_key", "sample_key.pub", "eKuiper")},
+			req:      httptest.NewRequest("GET", "http://127.0.0.1:9081/streams", nil),
+			res:      httptest.NewRecorder(),
+			wantCode: 200,
+		},
+
+		{
+			name:     "audience not right",
+			args:     args{th: genToken("sample_key", "sample_key.pub", "Neuron")},
+			req:      httptest.NewRequest("GET", "http://127.0.0.1:9081/streams", nil),
+			res:      httptest.NewRecorder(),
+			wantCode: 401,
+		},
+		{
+			name:     "no token",
+			args:     args{th: ""},
+			req:      httptest.NewRequest("GET", "http://127.0.0.1:9081/streams", nil),
+			res:      httptest.NewRecorder(),
+			wantCode: 401,
+		},
+		{
+			name:     "no need token path",
+			args:     args{th: ""},
+			req:      httptest.NewRequest("GET", "http://127.0.0.1:9081/ping", nil),
+			res:      httptest.NewRecorder(),
+			wantCode: 200,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.req.Header.Set("Authorization", tt.args.th)
+			handler.ServeHTTP(tt.res, tt.req)
+
+			res := tt.res.Result()
+
+			data, err := ioutil.ReadAll(res.Body)
+			if err != nil {
+				t.Errorf("expected error to be nil got %v", err)
+			}
+
+			if !reflect.DeepEqual(tt.wantCode, tt.res.Code) {
+				t.Errorf("expect %d, actual %d, result %s", tt.wantCode, tt.res.Code, string(data))
+			}
+
+			_ = res.Body.Close()
+		})
+	}
+}