|
@@ -2,7 +2,6 @@ package functions
|
|
|
|
|
|
import (
|
|
import (
|
|
"context"
|
|
"context"
|
|
- "crypto/ed25519"
|
|
|
|
"crypto/tls"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"crypto/x509"
|
|
"errors"
|
|
"errors"
|
|
@@ -23,7 +22,6 @@ import (
|
|
"github.com/gravitl/netmaker/netclient/daemon"
|
|
"github.com/gravitl/netmaker/netclient/daemon"
|
|
"github.com/gravitl/netmaker/netclient/ncutils"
|
|
"github.com/gravitl/netmaker/netclient/ncutils"
|
|
"github.com/gravitl/netmaker/netclient/wireguard"
|
|
"github.com/gravitl/netmaker/netclient/wireguard"
|
|
- ssl "github.com/gravitl/netmaker/tls"
|
|
|
|
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
|
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
|
)
|
|
)
|
|
|
|
|
|
@@ -71,7 +69,7 @@ func Daemon() error {
|
|
wg := sync.WaitGroup{}
|
|
wg := sync.WaitGroup{}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
wg.Add(1)
|
|
wg.Add(1)
|
|
- go Checkin(ctx, &wg)
|
|
|
|
|
|
+ go Checkin(ctx, &wg, serverSet)
|
|
quit := make(chan os.Signal, 1)
|
|
quit := make(chan os.Signal, 1)
|
|
signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
|
|
signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
|
|
<-quit
|
|
<-quit
|
|
@@ -172,107 +170,15 @@ func unsubscribeNode(client mqtt.Client, nodeCfg *config.ClientConfig) {
|
|
// the client should subscribe to ALL nodes that exist on server locally
|
|
// the client should subscribe to ALL nodes that exist on server locally
|
|
func messageQueue(ctx context.Context, server string) {
|
|
func messageQueue(ctx context.Context, server string) {
|
|
logger.Log(0, "netclient daemon started for server: ", server)
|
|
logger.Log(0, "netclient daemon started for server: ", server)
|
|
- client := setupMQTTSub(server)
|
|
|
|
|
|
+ client := setupMQTT(nil, server, false)
|
|
defer client.Disconnect(250)
|
|
defer client.Disconnect(250)
|
|
<-ctx.Done()
|
|
<-ctx.Done()
|
|
logger.Log(0, "shutting down daemon for server ", server)
|
|
logger.Log(0, "shutting down daemon for server ", server)
|
|
}
|
|
}
|
|
|
|
|
|
-// setupMQTTSub creates a connection to broker and subscribes to topic
|
|
|
|
-func setupMQTTSub(server string) mqtt.Client {
|
|
|
|
- opts := mqtt.NewClientOptions()
|
|
|
|
- opts.AddBroker("ssl://" + server + ":8883")
|
|
|
|
- opts.TLSConfig = NewTLSConfig(nil, server)
|
|
|
|
- opts.SetDefaultPublishHandler(All)
|
|
|
|
- opts.SetAutoReconnect(true)
|
|
|
|
- opts.SetConnectRetry(true)
|
|
|
|
- opts.SetConnectRetryInterval(time.Second << 2)
|
|
|
|
- opts.SetKeepAlive(time.Minute >> 1)
|
|
|
|
- opts.SetWriteTimeout(time.Minute)
|
|
|
|
- opts.SetOnConnectHandler(func(client mqtt.Client) {
|
|
|
|
- networks, err := ncutils.GetSystemNetworks()
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "error retriving networks ", err.Error())
|
|
|
|
- }
|
|
|
|
- for _, network := range networks {
|
|
|
|
- var currNodeCfg config.ClientConfig
|
|
|
|
- currNodeCfg.Network = network
|
|
|
|
- currNodeCfg.ReadConfig()
|
|
|
|
- if currNodeCfg.Server.Server == server {
|
|
|
|
- setSubscriptions(client, &currNodeCfg)
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- })
|
|
|
|
- opts.SetOrderMatters(true)
|
|
|
|
- opts.SetResumeSubs(true)
|
|
|
|
- opts.SetConnectionLostHandler(func(c mqtt.Client, e error) {
|
|
|
|
- logger.Log(0, "detected broker connection lost, running pull for all nodes")
|
|
|
|
- networks, err := ncutils.GetSystemNetworks()
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "error retriving networks ", err.Error())
|
|
|
|
- }
|
|
|
|
- for _, network := range networks {
|
|
|
|
- var cfg config.ClientConfig
|
|
|
|
- cfg.Network = network
|
|
|
|
- cfg.ReadConfig()
|
|
|
|
- _, err := Pull(cfg.Node.Network, true)
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "could not run pull, server unreachable: ", err.Error())
|
|
|
|
- logger.Log(0, "waiting to retry...")
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- // don't think following log message is accurate
|
|
|
|
- //logger.Log(0, "connection re-established with mqtt server")
|
|
|
|
- })
|
|
|
|
-
|
|
|
|
- client := mqtt.NewClient(opts)
|
|
|
|
- tperiod := time.Now().Add(12 * time.Second)
|
|
|
|
- for {
|
|
|
|
- //if after 12 seconds, try a pull on the last try
|
|
|
|
- if time.Now().After(tperiod) {
|
|
|
|
- networks, err := ncutils.GetSystemNetworks()
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "error retriving networks ", err.Error())
|
|
|
|
- }
|
|
|
|
- for _, network := range networks {
|
|
|
|
- var cfg config.ClientConfig
|
|
|
|
- cfg.Network = network
|
|
|
|
- cfg.ReadConfig()
|
|
|
|
- if cfg.Server.Server == server {
|
|
|
|
- _, err := Pull(cfg.Node.Network, true)
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "could not run pull, exiting ", cfg.Node.Network, " setup: ", err.Error())
|
|
|
|
- return client
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- time.Sleep(time.Second)
|
|
|
|
- }
|
|
|
|
- if token := client.Connect(); token.Wait() && token.Error() != nil {
|
|
|
|
- logger.Log(0, "unable to connect to broker, retrying ...")
|
|
|
|
- if time.Now().After(tperiod) {
|
|
|
|
- logger.Log(0, "could not connect to broker, exiting ", server, " setup: ", token.Error().Error())
|
|
|
|
- if strings.Contains(token.Error().Error(), "connectex") || strings.Contains(token.Error().Error(), "i/o timeout") {
|
|
|
|
- logger.Log(0, "connection issue detected.. restarting daemon")
|
|
|
|
- daemon.Restart()
|
|
|
|
- }
|
|
|
|
- return client
|
|
|
|
- }
|
|
|
|
- } else {
|
|
|
|
- break
|
|
|
|
- }
|
|
|
|
- time.Sleep(2 * time.Second)
|
|
|
|
- }
|
|
|
|
- return client
|
|
|
|
-}
|
|
|
|
-
|
|
|
|
// NewTLSConf sets up tls configuration to connect to broker securely
|
|
// NewTLSConf sets up tls configuration to connect to broker securely
|
|
-func NewTLSConfig(cfg *config.ClientConfig, server string) *tls.Config {
|
|
|
|
- var file string
|
|
|
|
- if cfg != nil {
|
|
|
|
- server = cfg.Server.Server
|
|
|
|
- }
|
|
|
|
- file = ncutils.GetNetclientServerPath(server) + "/root.pem"
|
|
|
|
|
|
+func NewTLSConfig(server string) *tls.Config {
|
|
|
|
+ file := ncutils.GetNetclientServerPath(server) + "/root.pem"
|
|
certpool := x509.NewCertPool()
|
|
certpool := x509.NewCertPool()
|
|
ca, err := os.ReadFile(file)
|
|
ca, err := os.ReadFile(file)
|
|
if err != nil {
|
|
if err != nil {
|
|
@@ -288,42 +194,24 @@ func NewTLSConfig(cfg *config.ClientConfig, server string) *tls.Config {
|
|
}
|
|
}
|
|
certs := []tls.Certificate{clientKeyPair}
|
|
certs := []tls.Certificate{clientKeyPair}
|
|
return &tls.Config{
|
|
return &tls.Config{
|
|
- RootCAs: certpool,
|
|
|
|
- ClientAuth: tls.NoClientCert,
|
|
|
|
- ClientCAs: nil,
|
|
|
|
- Certificates: certs,
|
|
|
|
- //InsecureSkipVerify: false fails ---- so need to use VerifyConnection
|
|
|
|
- InsecureSkipVerify: true,
|
|
|
|
- VerifyConnection: func(cs tls.ConnectionState) error {
|
|
|
|
- if cs.ServerName != server {
|
|
|
|
- logger.Log(0, "VerifyConnection - certifiate mismatch")
|
|
|
|
- return errors.New("certificate doesn't match server")
|
|
|
|
- }
|
|
|
|
- ca, err := ssl.ReadCert(ncutils.GetNetclientServerPath(cs.ServerName) + "/root.pem")
|
|
|
|
- if err != nil {
|
|
|
|
- logger.Log(0, "VerifyConnection - unable to read ca", err.Error())
|
|
|
|
- return errors.New("unable to read ca")
|
|
|
|
- }
|
|
|
|
- for _, cert := range cs.PeerCertificates {
|
|
|
|
- if cert.IsCA {
|
|
|
|
- if string(cert.PublicKey.(ed25519.PublicKey)) != string(ca.PublicKey.(ed25519.PublicKey)) {
|
|
|
|
- logger.Log(0, "VerifyConnection - public key mismatch")
|
|
|
|
- return errors.New("cert public key does not match ca public key")
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- }
|
|
|
|
- return nil
|
|
|
|
- },
|
|
|
|
|
|
+ RootCAs: certpool,
|
|
|
|
+ ClientAuth: tls.NoClientCert,
|
|
|
|
+ ClientCAs: nil,
|
|
|
|
+ Certificates: certs,
|
|
|
|
+ InsecureSkipVerify: false,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
-// setupMQTT creates a connection to broker and return client
|
|
|
|
|
|
+// setupMQTT creates a connection to broker and returns client
|
|
// this function is primarily used to create a connection to publish to the broker
|
|
// this function is primarily used to create a connection to publish to the broker
|
|
-func setupMQTT(cfg *config.ClientConfig, publish bool) mqtt.Client {
|
|
|
|
|
|
+func setupMQTT(cfg *config.ClientConfig, server string, publish bool) mqtt.Client {
|
|
opts := mqtt.NewClientOptions()
|
|
opts := mqtt.NewClientOptions()
|
|
- server := cfg.Server.Server
|
|
|
|
- opts.AddBroker("ssl://" + server + ":8883")
|
|
|
|
- opts.TLSConfig = NewTLSConfig(cfg, "")
|
|
|
|
|
|
+ if cfg != nil {
|
|
|
|
+ server = cfg.Server.Server
|
|
|
|
+ }
|
|
|
|
+ opts.AddBroker("ssl://" + server + ":8883") // TODO get the appropriate port of the comms mq server
|
|
|
|
+ opts.SetTLSConfig(NewTLSConfig(server))
|
|
|
|
+ opts.SetClientID(ncutils.MakeRandomString(23))
|
|
opts.SetDefaultPublishHandler(All)
|
|
opts.SetDefaultPublishHandler(All)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetConnectRetry(true)
|
|
opts.SetConnectRetry(true)
|