Protocol: renew device certificate if it is not valid

The device certificate was obtained only once, during the first connection. However, certificates may change over time
(e.g. by substituting an expired one with a valid one). Obtain a new certificate if the old one is not valid anymore.

As paho mqtt client does not expose TLS errors upon reconnection, a custom open connection function was added.

Signed-off-by: Arnaldo Cesco <[email protected]>

device/device.go

@@ -221,6 +221,8 @@ func (d *Device) Connect(result chan<- error) {
 		}
 
 		// Wait for the token - we're in a coroutine anyway
+		// If AutoReconnect is enabled, reconnections will be handled by
+		// paho's autoreconnect mechanism.
 		policy.Reset()
 		connectOperation := func() error {
 			connectToken := d.m.Connect().(*mqtt.ConnectToken)

device/protocol_mqtt_v1.go

@@ -17,10 +17,14 @@ package device
 import (
 	"bytes"
 	"compress/zlib"
+	"crypto/tls"
 	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
+	"net"
+	"net/url"
+	"os"
 	"path/filepath"
 	"reflect"
 	"sort"
@@ -31,6 +35,7 @@ import (
 	mqtt "github.com/ispirata/paho.mqtt.golang"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"golang.org/x/net/proxy"
 )
 
 func (d *Device) getBaseTopic() string {
@@ -50,11 +55,8 @@ func (d *Device) initializeMQTTClient() error {
 		opts.SetStore(s)
 	}
 
-	tlsConfig, err := d.getTLSConfig()
-	if err != nil {
-		return err
-	}
-	opts.SetTLSConfig(tlsConfig)
+	// remove TLS config as it will be handled in the custom connection function
+	opts.SetCustomOpenConnectionFn(d.astarteCustomOpenConnectionFn)
 
 	opts.SetOnConnectHandler(func(client mqtt.Client, sessionPresent bool) {
 		astarteOnConnectHandler(d, sessionPresent)
@@ -203,6 +205,65 @@ func (d *Device) handleControlMessages(message string, payload []byte) error {
 	return nil
 }
 
+// TLS errors on reconnection are not exposed, so we have to make do.
+// This code is mostly taken from Paho's openConnection function. It does not carry out any MQTT specific handshakes.
+func (d *Device) astarteCustomOpenConnectionFn(uri *url.URL, options mqtt.ClientOptions) (net.Conn, error) {
+	switch uri.Scheme {
+	case "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
+		// use our own TLS config as the one in mqtt.options might be outdated
+		tlsc, tlscErr := d.getTLSConfig()
+		if tlscErr != nil {
+			return nil, tlscErr
+		}
+		allProxy := os.Getenv("all_proxy")
+		if len(allProxy) == 0 {
+			//Do not use a backoff, as this code might be called many consecutive times until a valid device certificate is emitted.
+			dialer := &net.Dialer{Timeout: 10 * time.Second}
+			conn, err := tls.DialWithDialer(dialer, "tcp", uri.Host, tlsc)
+			if err != nil {
+				_ = d.obtainNewCertificate()
+				//fail anyway, so the next reconnection will use the new certificate.
+				return nil, err
+			}
+			return conn, nil
+		}
+		proxyDialer := proxy.FromEnvironment()
+		conn, err := proxyDialer.Dial("tcp", uri.Host)
+		if err != nil {
+			return nil, err
+		}
+		tlsConn := tls.Client(conn, tlsc)
+		err = tlsConn.Handshake()
+		if err != nil {
+			_ = conn.Close()
+			_ = d.obtainNewCertificate()
+			//fail anyway, so the next reconnection will use the new certificate.
+			return nil, err
+		}
+		return tlsConn, nil
+
+	//no need for certificates
+	case "mqtt", "tcp":
+		allProxy := os.Getenv("all_proxy")
+		if len(allProxy) == 0 {
+			dialer := &net.Dialer{Timeout: 10 * time.Second}
+			conn, err := dialer.Dial("tcp", uri.Host)
+			if err != nil {
+				return nil, err
+			}
+			return conn, nil
+		}
+		proxyDialer := proxy.FromEnvironment()
+
+		conn, err := proxyDialer.Dial("tcp", uri.Host)
+		if err != nil {
+			return nil, err
+		}
+		return conn, nil
+	}
+	return nil, errors.New("unknown protocol")
+}
+
 func astarteOnConnectHandler(d *Device, sessionPresent bool) {
 	// Generate Introspection first
 	introspection := d.generateDeviceIntrospection()