add insecureSkipVerify conf

docs/en_US/rules/sinks/mqtt.md

@@ -2,17 +2,18 @@
 
 The action is used for publish output message into a MQTT server. 
 
-| Property name     | Optional | Description                                                  |
-| ----------------- | -------- | ------------------------------------------------------------ |
-| server            | false    | The broker address of the mqtt server, such as ``tcp://127.0.0.1:1883`` |
-| topic             | false    | The mqtt topic, such as ``analysis/result``                  |
-| clientId          | true     | The client id for mqtt connection. If not specified, an uuid will be used |
-| protocolVersion   | true     | 3.1 (also refer as MQTT 3) or 3.1.1 (also refer as MQTT 4).  If not specified, the default value is 3.1. |
-| qos               | true     | The QoS for message delivery.                                |
-| username          | true     | The user name for the connection.                            |
-| password          | true     | The password for the connection.                             |
-| certificationPath | true     | The certification path. It can be an absolute path, or a relative path. If it is an relative path, then the base path is where you excuting the ``server`` command. For example, if you run ``bin/server`` from ``/var/kuiper``, then the base path is ``/var/kuiper``; If you run ``./server`` from ``/var/kuiper/bin``, then the base path is ``/var/kuiper/bin``. |
-| privateKeyPath    | true     | The private key path. It can be either absolute path, or relative path. For more detailed information, please refer to ``certificationPath``. |
+| Property name      | Optional | Description                                                  |
+| ------------------ | -------- | ------------------------------------------------------------ |
+| server             | false    | The broker address of the mqtt server, such as ``tcp://127.0.0.1:1883`` |
+| topic              | false    | The mqtt topic, such as ``analysis/result``                  |
+| clientId           | true     | The client id for mqtt connection. If not specified, an uuid will be used |
+| protocolVersion    | true     | 3.1 (also refer as MQTT 3) or 3.1.1 (also refer as MQTT 4).  If not specified, the default value is 3.1. |
+| qos                | true     | The QoS for message delivery.                                |
+| username           | true     | The user name for the connection.                            |
+| password           | true     | The password for the connection.                             |
+| certificationPath  | true     | The certification path. It can be an absolute path, or a relative path. If it is an relative path, then the base path is where you excuting the ``server`` command. For example, if you run ``bin/server`` from ``/var/kuiper``, then the base path is ``/var/kuiper``; If you run ``./server`` from ``/var/kuiper/bin``, then the base path is ``/var/kuiper/bin``. |
+| privateKeyPath     | true     | The private key path. It can be either absolute path, or relative path. For more detailed information, please refer to ``certificationPath``. |
+| insecureSkipVerify | true     | If InsecureSkipVerify is ``true``, TLS accepts any certificate presented by the server and any host name in that certificate.  In this mode, TLS is susceptible to man-in-the-middle attacks. The default value is ``false``. The configuration item can only be used with TLS connections. |
 
 Below is sample configuration for connecting to Azure IoT Hub by using SAS authentication.
 ```json
@@ -39,7 +40,8 @@ Below is another sample configuration for connecting to AWS IoT by using certifi
         "qos": 1,
         "clientId": "demo_001",
         "certificationPath": "keys/d3807d9fa5-certificate.pem",
-        "privateKeyPath": "keys/d3807d9fa5-private.pem.key"
+        "privateKeyPath": "keys/d3807d9fa5-private.pem.key", 
+        "insecureSkipVerify": false
       }
     }
 ```

xstream/sinks/mqtt_sink.go

@@ -20,6 +20,8 @@ type MQTTSink struct {
 	certPath string
 	pkeyPath string
 
+	insecureSkipVerify bool
+
 	conn MQTT.Client
 }
 
@@ -86,6 +88,13 @@ func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
 		}
 	}
 
+	insecureSkipVerify := false
+	if pk, ok := ps["insecureSkipVerify"]; ok {
+		if v, ok := pk.(bool); ok {
+			insecureSkipVerify = v
+		}
+	}
+
 	ms.srv = srv.(string)
 	ms.tpc = tpc.(string)
 	ms.clientid = clientid.(string)
@@ -94,6 +103,7 @@ func (ms *MQTTSink) Configure(ps map[string]interface{}) error {
 	ms.password = password
 	ms.certPath = certPath
 	ms.pkeyPath = pKeyPath
+	ms.insecureSkipVerify = insecureSkipVerify
 
 	return nil
 }
@@ -110,7 +120,7 @@ func (ms *MQTTSink) Open(ctx api.StreamContext) error {
 				if cer, err2 := tls.LoadX509KeyPair(cp, kp); err2 != nil {
 					return err2
 				} else {
-					opts.SetTLSConfig(&tls.Config{Certificates: []tls.Certificate{cer}})
+					opts.SetTLSConfig(&tls.Config{Certificates: []tls.Certificate{cer}, InsecureSkipVerify: ms.insecureSkipVerify})
 				}
 			} else {
 				return err1