Merge branch 'develop' into feature_v0.10.0_initialize_iptables

.gitignore

@@ -17,3 +17,4 @@ config/dnsconfig/
 data/
 .vscode/
 .idea/
+

compose/docker-compose.contained.yml

@@ -3,7 +3,7 @@ version: "3.4"
 services:
   netmaker:
     container_name: netmaker
-    image: gravitl/netmaker:v0.10.0.12
+    image: gravitl/netmaker:v0.10.0
     volumes:
       - dnsconfig:/root/config/dnsconfig
       - /usr/bin/wg:/usr/bin/wg
@@ -72,9 +72,20 @@ services:
       # - $PWD/site:/srv # you could also serve a static site in site folder
       - caddy_data:/data
       - caddy_conf:/config
+  mosquitto:
+    image: eclipse-mosquitto:2.0.14
+    container_name: broker
+    restart: unless-stopped
+    network_mode: host
+    volumes:
+      - /root/mosquitto.conf:/mosquitto/config/mosquitto.conf
+      - mosquitto_data:/mosquitto/data
+      - mosquitto_logs:/mosquitto/log 
 volumes:
   caddy_data: {}
   caddy_conf: {}
   sqldata: {}
   dnsconfig: {}
+  mosquitto_data: {}
+  mosquitto_logs: {}
 

config/config.go

@@ -43,12 +43,14 @@ type ServerConfig struct {
 	GRPCHost              string `yaml:"grpchost"`
 	GRPCPort              string `yaml:"grpcport"`
 	GRPCSecure            string `yaml:"grpcsecure"`
+	MQHOST                string `yaml:"mqhost"`
 	MasterKey             string `yaml:"masterkey"`
 	DNSKey                string `yaml:"dnskey"`
 	AllowedOrigin         string `yaml:"allowedorigin"`
 	NodeID                string `yaml:"nodeid"`
 	RestBackend           string `yaml:"restbackend"`
 	AgentBackend          string `yaml:"agentbackend"`
+	MessageQueueBackend   string `yaml:"messagequeuebackend"`
 	ClientMode            string `yaml:"clientmode"`
 	DNSMode               string `yaml:"dnsmode"`
 	SplitDNS              string `yaml:"splitdns"`
@@ -70,6 +72,7 @@ type ServerConfig struct {
 	DisplayKeys           string `yaml:"displaykeys"`
 	AzureTenant           string `yaml:"azuretenant"`
 	RCE                   string `yaml:"rce"`
+	Debug                 bool   `yaml:"debug"`
 	Telemetry             string `yaml:"telemetry"`
 	ManageIPTables        string `yaml:"manageiptables"`
 	PortForwardServices   string `yaml:"portforwardservices"`

controllers/config/dnsconfig/netmaker.hosts

@@ -0,0 +1,2 @@
+10.0.0.1         testnode.skynet
+10.0.0.2         myhost.skynet

controllers/controller.go

@@ -15,6 +15,7 @@ import (
 	"github.com/gravitl/netmaker/servercfg"
 )
 
+// HttpHandlers - handler functions for REST interactions
 var HttpHandlers = []interface{}{
 	nodeHandlers,
 	userHandlers,
@@ -23,7 +24,6 @@ var HttpHandlers = []interface{}{
 	fileHandlers,
 	serverHandlers,
 	extClientHandlers,
-	loggerHandlers,
 }
 
 // HandleRESTRequests - handles the rest requests
@@ -64,7 +64,7 @@ func HandleRESTRequests(wg *sync.WaitGroup) {
 
 	// After receiving CTRL+C Properly stop the server
 	logger.Log(0, "Stopping the REST server...")
-	srv.Shutdown(context.TODO())
 	logger.Log(0, "REST Server closed.")
 	logger.DumpFile(fmt.Sprintf("data/netmaker.log.%s", time.Now().Format(logger.TimeFormatDay)))
+	srv.Shutdown(context.TODO())
 }

controllers/logger.go

@@ -1,23 +0,0 @@
-package controller
-
-import (
-	"fmt"
-	"net/http"
-	"time"
-
-	"github.com/gorilla/mux"
-	"github.com/gravitl/netmaker/logger"
-)
-
-func loggerHandlers(r *mux.Router) {
-	r.HandleFunc("/api/logs", securityCheck(true, http.HandlerFunc(getLogs))).Methods("GET")
-}
-
-func getLogs(w http.ResponseWriter, r *http.Request) {
-	var currentTime = time.Now().Format(logger.TimeFormatDay)
-	var currentFilePath = fmt.Sprintf("data/netmaker.log.%s", currentTime)
-	logger.DumpFile(currentFilePath)
-	logger.ResetLogs()
-	w.WriteHeader(http.StatusOK)
-	w.Write([]byte(logger.Retrieve(currentFilePath)))
-}

controllers/network.go

@@ -182,8 +182,8 @@ func updateNetworkNodeLimit(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(network)
 }
 
-//Delete a network
-//Will stop you if  there's any nodes associated
+// Delete a network
+// Will stop you if  there's any nodes associated
 func deleteNetwork(w http.ResponseWriter, r *http.Request) {
 	// Set header
 	w.Header().Set("Content-Type", "application/json")

controllers/node.go

@@ -11,6 +11,7 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/mq"
 	"github.com/gravitl/netmaker/servercfg"
 	"golang.org/x/crypto/bcrypt"
 )
@@ -32,6 +33,7 @@ func nodeHandlers(r *mux.Router) {
 	r.HandleFunc("/api/nodes/{network}", createNode).Methods("POST")
 	r.HandleFunc("/api/nodes/adm/{network}/lastmodified", authorize(true, "network", http.HandlerFunc(getLastModified))).Methods("GET")
 	r.HandleFunc("/api/nodes/adm/{network}/authenticate", authenticate).Methods("POST")
+
 }
 
 func authenticate(response http.ResponseWriter, request *http.Request) {
@@ -427,6 +429,14 @@ func uncordonNode(w http.ResponseWriter, r *http.Request) {
 	if err = runServerPeerUpdate(node.Network, false); err != nil {
 		logger.Log(1, "internal error when approving node:", nodeid)
 	}
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "uncordoned node", node.Name)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode("SUCCESS")
@@ -451,6 +461,14 @@ func createEgressGateway(w http.ResponseWriter, r *http.Request) {
 	if err = runServerPeerUpdate(gateway.NetID, true); err != nil {
 		logger.Log(1, "internal error when setting peers after creating egress on node:", gateway.NodeID)
 	}
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update "+err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "created egress gateway on node", gateway.NodeID, "on network", gateway.NetID)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)
@@ -469,6 +487,14 @@ func deleteEgressGateway(w http.ResponseWriter, r *http.Request) {
 	if err = runServerPeerUpdate(netid, true); err != nil {
 		logger.Log(1, "internal error when setting peers after removing egress on node:", nodeid)
 	}
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "deleted egress gateway", nodeid, "on network", netid)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)
@@ -486,7 +512,14 @@ func createIngressGateway(w http.ResponseWriter, r *http.Request) {
 		returnErrorResponse(w, r, formatError(err, "internal"))
 		return
 	}
-
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "created ingress gateway on node", nodeid, "on network", netid)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)
@@ -501,7 +534,14 @@ func deleteIngressGateway(w http.ResponseWriter, r *http.Request) {
 		returnErrorResponse(w, r, formatError(err, "internal"))
 		return
 	}
-
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "deleted ingress gateway", nodeid)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)
@@ -553,7 +593,6 @@ func updateNode(w http.ResponseWriter, r *http.Request) {
 		returnErrorResponse(w, r, formatError(err, "internal"))
 		return
 	}
-
 	if relayupdate {
 		logic.UpdateRelay(node.Network, node.RelayAddrs, newNode.RelayAddrs)
 		if err = logic.NetworkNodesUpdatePullChanges(node.Network); err != nil {
@@ -561,7 +600,7 @@ func updateNode(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	if servercfg.IsDNSMode() { // TODO check when this should be updated..
+	if servercfg.IsDNSMode() {
 		err = logic.SetDNS()
 	}
 
@@ -570,9 +609,19 @@ func updateNode(w http.ResponseWriter, r *http.Request) {
 		returnErrorResponse(w, r, formatError(err, "internal"))
 		return
 	}
-	logger.Log(1, r.Header.Get("user"), "updated node", node.ID)
+	logger.Log(1, r.Header.Get("user"), "updated node", node.MacAddress, "on network", node.Network)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(newNode)
+	go func() {
+		if err := mq.NodeUpdate(&newNode); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if logic.ShouldPeersUpdate(&node, &newNode) {
+			if err := mq.UpdatePeers(&newNode); err != nil {
+				logger.Log(1, "error publishing peer update after node update", err.Error())
+			}
+		}
+	}()
 }
 
 func deleteNode(w http.ResponseWriter, r *http.Request) {
@@ -598,7 +647,12 @@ func deleteNode(w http.ResponseWriter, r *http.Request) {
 		returnErrorResponse(w, r, formatError(err, "internal"))
 		return
 	}
-
+	node.Action = models.NODE_DELETE
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node delete ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "Deleted node", nodeid, "from network", params["network"])
 	returnSuccessResponse(w, r, nodeid+" deleted.")
 }

controllers/node_grpc.go

@@ -10,6 +10,7 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/mq"
 	"github.com/gravitl/netmaker/servercfg"
 )
 
@@ -85,12 +86,17 @@ func (s *NodeServiceServer) CreateNode(ctx context.Context, req *nodepb.Object)
 	if err != nil {
 		return nil, err
 	}
-
 	err = runServerPeerUpdate(node.Network, true)
 	if err != nil {
 		logger.Log(1, "internal error when setting peers after node,", node.ID, "was created (gRPC)")
 	}
 	logger.Log(0, "new node,", node.Name, ", added on network,"+node.Network)
+	// notify other nodes on network of new peer
+	go func() {
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(0, "failed to inform peers of new node ", err.Error())
+		}
+	}()
 
 	return response, nil
 }
@@ -148,11 +154,16 @@ func (s *NodeServiceServer) DeleteNode(ctx context.Context, req *nodepb.Object)
 	if err != nil {
 		return nil, err
 	}
-
 	err = runServerPeerUpdate(node.Network, true)
 	if err != nil {
 		logger.Log(1, "internal error when setting peers after deleting node:", node.ID, "over gRPC")
 	}
+	// notify other nodes on network of deleted peer
+	go func() {
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(0, "failed to inform peers of deleted node ", err.Error())
+		}
+	}()
 
 	return &nodepb.Object{
 		Data: "success",

controllers/node_test.go

@@ -79,13 +79,12 @@ func TestGetNetworkNodes(t *testing.T) {
 	t.Run("BadNet", func(t *testing.T) {
 		node, err := logic.GetNetworkNodes("badnet")
 		assert.Nil(t, err)
-		assert.Equal(t, []models.Node{}, node)
-		//assert.Equal(t, "mongo: no documents in result", err.Error())
+		assert.Nil(t, node)
 	})
 	t.Run("NoNodes", func(t *testing.T) {
 		node, err := logic.GetNetworkNodes("skynet")
 		assert.Nil(t, err)
-		assert.Equal(t, []models.Node{}, node)
+		assert.Nil(t, node)
 	})
 	t.Run("Success", func(t *testing.T) {
 		createTestNode()

controllers/relay.go

@@ -8,6 +8,7 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/mq"
 )
 
 func createRelay(w http.ResponseWriter, r *http.Request) {
@@ -29,6 +30,14 @@ func createRelay(w http.ResponseWriter, r *http.Request) {
 	if err = runServerPeerUpdate(relay.NetID, true); err != nil {
 		logger.Log(1, "internal error when creating relay on node:", relay.NodeID)
 	}
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "created relay on node", relay.NodeID, "on network", relay.NetID)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)
@@ -47,6 +56,14 @@ func deleteRelay(w http.ResponseWriter, r *http.Request) {
 	if err = runServerPeerUpdate(netid, true); err != nil {
 		logger.Log(1, "internal error when deleting relay on node:", nodeid)
 	}
+	go func() {
+		if err := mq.NodeUpdate(&node); err != nil {
+			logger.Log(1, "error publishing node update", err.Error())
+		}
+		if err := mq.UpdatePeers(&node); err != nil {
+			logger.Log(1, "error publishing peer update ", err.Error())
+		}
+	}()
 	logger.Log(1, r.Header.Get("user"), "deleted relay server", nodeid, "on network", netid)
 	w.WriteHeader(http.StatusOK)
 	json.NewEncoder(w).Encode(node)

controllers/server_util.go

@@ -4,22 +4,19 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/servercfg"
-	"github.com/gravitl/netmaker/serverctl"
 )
 
 func runServerPeerUpdate(network string, shouldPeerUpdate bool) error {
-	if servercfg.Telemetry() == "on" {
-		err := serverctl.TelemetryCheckpoint()
-		if err != nil {
-			logger.Log(1, "failed to send telemetry:", err.Error())
-		}
+	err := logic.TimerCheckpoint()
+	if err != nil {
+		logger.Log(3, "error occurred on timer,", err.Error())
 	}
 	if servercfg.IsClientMode() != "on" {
 		return nil
 	}
-	var currentServerNodeID, err = logic.GetNetworkServerNodeID(network)
+	var currentServerNodeID, getErr = logic.GetNetworkServerNodeID(network)
 	if err != nil {
-		return err
+		return getErr
 	}
 	var currentServerNode, currErr = logic.GetNodeByID(currentServerNodeID)
 	if currErr != nil {

database/database.go

@@ -206,7 +206,7 @@ func initializeUUID() error {
 		return nil
 	}
 	telemetry := models.Telemetry{UUID: uuid.NewString()}
-	telJSON, err := json.Marshal(telemetry)
+	telJSON, err := json.Marshal(&telemetry)
 	if err != nil {
 		return err
 	}

docker/mosquitto.conf

@@ -0,0 +1,10 @@
+persistence true
+persistence file mosquitto.db
+persistance_location mosquitto/data
+
+log_dest_file mosquitto/log/mosquitto.log
+
+per_listener_settings true
+listener 1883
+allow_anonymous true
+

go.mod

@@ -3,9 +3,11 @@ module github.com/gravitl/netmaker
 go 1.17
 
 require (
+	github.com/eclipse/paho.mqtt.golang v1.3.5
 	github.com/go-playground/validator/v10 v10.10.0
 	github.com/golang-jwt/jwt/v4 v4.2.0
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/uuid v1.3.0
 	github.com/gorilla/handlers v1.5.1
 	github.com/gorilla/mux v1.8.0
 	github.com/lib/pq v1.10.4
@@ -25,6 +27,7 @@ require (
 	google.golang.org/genproto v0.0.0-20210201151548-94839c025ad4 // indirect
 	google.golang.org/grpc v1.43.0
 	google.golang.org/protobuf v1.27.1
+	gopkg.in/ini.v1 v1.66.2
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 )
 
@@ -36,7 +39,7 @@ require (
 	github.com/go-playground/locales v0.14.0 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/google/go-cmp v0.5.5 // indirect
-	github.com/google/uuid v1.3.0 // indirect
+	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/josharian/native v0.0.0-20200817173448-b6b71def0850 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/mdlayher/genetlink v1.0.0 // indirect

go.sum

@@ -24,6 +24,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 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/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/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -76,6 +78,8 @@ github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH
 github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
@@ -185,6 +189,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191007182048-72f939374954/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
@@ -286,6 +291,8 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
+gopkg.in/ini.v1 v1.66.2 h1:XfR1dOYubytKy4Shzc2LHrrGhU0lDCfDGG1yLPmpgsI=
+gopkg.in/ini.v1 v1.66.2/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

logic/nodes.go

@@ -20,11 +20,12 @@ import (
 
 // GetNetworkNodes - gets the nodes of a network
 func GetNetworkNodes(network string) ([]models.Node, error) {
-	var nodes, err = GetAllNodes()
+	var nodes []models.Node
+	allnodes, err := GetAllNodes()
 	if err != nil {
 		return []models.Node{}, err
 	}
-	for _, node := range nodes {
+	for _, node := range allnodes {
 		if node.Network == network {
 			nodes = append(nodes, node)
 		}
@@ -244,6 +245,7 @@ func ShouldPeersUpdate(currentNode *models.Node, newNode *models.Node) bool {
 		newNode.IsRelay != currentNode.IsRelay ||
 		newNode.UDPHolePunch != currentNode.UDPHolePunch ||
 		newNode.IsPending != currentNode.IsPending ||
+		newNode.PersistentKeepalive != currentNode.PersistentKeepalive ||
 		len(newNode.ExcludedAddrs) != len(currentNode.ExcludedAddrs) ||
 		len(newNode.AllowedIPs) != len(currentNode.AllowedIPs) {
 		return true

logic/peers.go

@@ -0,0 +1,124 @@
+package logic
+
+import (
+	"log"
+	"net"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/netclient/ncutils"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+// GetPeerUpdate - gets a wireguard peer config for each peer of a node
+func GetPeerUpdate(node *models.Node) (models.PeerUpdate, error) {
+	var peerUpdate models.PeerUpdate
+	var peers []wgtypes.PeerConfig
+	networkNodes, err := GetNetworkNodes(node.Network)
+	if err != nil {
+		return models.PeerUpdate{}, err
+	}
+	for _, peer := range networkNodes {
+		if peer.ID == node.ID {
+			//skip yourself
+			continue
+		}
+		pubkey, err := wgtypes.ParseKey(peer.PublicKey)
+		if err != nil {
+			return models.PeerUpdate{}, err
+		}
+		if node.Endpoint == peer.Endpoint {
+			//peer is on same network
+			if node.LocalAddress != peer.LocalAddress && peer.LocalAddress != "" {
+				peer.Endpoint = peer.LocalAddress
+			} else {
+				continue
+			}
+		}
+		endpoint := peer.Endpoint + ":" + strconv.FormatInt(int64(peer.ListenPort), 10)
+		address, err := net.ResolveUDPAddr("udp", endpoint)
+		if err != nil {
+			return models.PeerUpdate{}, err
+		}
+		allowedips := GetAllowedIPs(node, &peer)
+		var keepalive time.Duration
+		if node.PersistentKeepalive != 0 {
+			keepalive, _ = time.ParseDuration(strconv.FormatInt(int64(node.PersistentKeepalive), 10) + "s")
+		}
+		var peerData = wgtypes.PeerConfig{
+			PublicKey:                   pubkey,
+			Endpoint:                    address,
+			ReplaceAllowedIPs:           true,
+			AllowedIPs:                  allowedips,
+			PersistentKeepaliveInterval: &keepalive,
+		}
+		peers = append(peers, peerData)
+	}
+	peerUpdate.Network = node.Network
+	peerUpdate.Peers = peers
+	return peerUpdate, nil
+}
+
+// GetAllowedIPs - calculates the wireguard allowedip field for a peer of a node based on the peer and node settings
+func GetAllowedIPs(node, peer *models.Node) []net.IPNet {
+	var allowedips []net.IPNet
+	var gateways []string
+	var peeraddr = net.IPNet{
+		IP:   net.ParseIP(peer.Address),
+		Mask: net.CIDRMask(32, 32),
+	}
+	dualstack := false
+	allowedips = append(allowedips, peeraddr)
+	// handle manually set peers
+	for _, allowedIp := range node.AllowedIPs {
+		if _, ipnet, err := net.ParseCIDR(allowedIp); err == nil {
+			nodeEndpointArr := strings.Split(node.Endpoint, ":")
+			if !ipnet.Contains(net.IP(nodeEndpointArr[0])) && ipnet.IP.String() != node.Address { // don't need to add an allowed ip that already exists..
+				allowedips = append(allowedips, *ipnet)
+			}
+		} else if appendip := net.ParseIP(allowedIp); appendip != nil && allowedIp != node.Address {
+			ipnet := net.IPNet{
+				IP:   net.ParseIP(allowedIp),
+				Mask: net.CIDRMask(32, 32),
+			}
+			allowedips = append(allowedips, ipnet)
+		}
+	}
+	// handle egress gateway peers
+	if node.IsEgressGateway == "yes" {
+		//hasGateway = true
+		ranges := node.EgressGatewayRanges
+		for _, iprange := range ranges { // go through each cidr for egress gateway
+			_, ipnet, err := net.ParseCIDR(iprange) // confirming it's valid cidr
+			if err != nil {
+				ncutils.PrintLog("could not parse gateway IP range. Not adding "+iprange, 1)
+				continue // if can't parse CIDR
+			}
+			nodeEndpointArr := strings.Split(node.Endpoint, ":") // getting the public ip of node
+			if ipnet.Contains(net.ParseIP(nodeEndpointArr[0])) { // ensuring egress gateway range does not contain public ip of node
+				ncutils.PrintLog("egress IP range of "+iprange+" overlaps with "+node.Endpoint+", omitting", 2)
+				continue // skip adding egress range if overlaps with node's ip
+			}
+			if ipnet.Contains(net.ParseIP(node.LocalAddress)) { // ensuring egress gateway range does not contain public ip of node
+				ncutils.PrintLog("egress IP range of "+iprange+" overlaps with "+node.LocalAddress+", omitting", 2)
+				continue // skip adding egress range if overlaps with node's local ip
+			}
+			gateways = append(gateways, iprange)
+			if err != nil {
+				log.Println("ERROR ENCOUNTERED SETTING GATEWAY")
+			} else {
+				allowedips = append(allowedips, *ipnet)
+			}
+		}
+	}
+	if node.Address6 != "" && dualstack {
+		var addr6 = net.IPNet{
+			IP:   net.ParseIP(node.Address6),
+			Mask: net.CIDRMask(128, 128),
+		}
+		allowedips = append(allowedips, addr6)
+	}
+	return allowedips
+}

logic/server.go

@@ -17,7 +17,6 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
-// == Public ==
 // == Join, Checkin, and Leave for Server ==
 
 // KUBERNETES_LISTEN_PORT - starting port for Kubernetes in order to use NodePort range

serverctl/telemetry.go → logic/telemetry.go

@@ -1,68 +1,45 @@
-package serverctl
+package logic
 
 import (
 	"encoding/json"
 	"time"
 
 	"github.com/gravitl/netmaker/database"
-	"github.com/gravitl/netmaker/logger"
-	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/models"
 	"github.com/gravitl/netmaker/servercfg"
 	"github.com/posthog/posthog-go"
 )
 
-// POSTHOG_PUB_KEY - Key for sending data to PostHog
-const POSTHOG_PUB_KEY = "phc_1vEXhPOA1P7HP5jP2dVU9xDTUqXHAelmtravyZ1vvES"
+// posthog_pub_key - Key for sending data to PostHog
+const posthog_pub_key = "phc_1vEXhPOA1P7HP5jP2dVU9xDTUqXHAelmtravyZ1vvES"
 
-// POSTHOG_ENDPOINT - Endpoint of PostHog server
-const POSTHOG_ENDPOINT = "https://app.posthog.com"
+// posthog_endpoint - Endpoint of PostHog server
+const posthog_endpoint = "https://app.posthog.com"
 
-// TELEMETRY_HOURS_BETWEEN_SEND - How long to wait before sending telemetry to server (24 hours)
-const TELEMETRY_HOURS_BETWEEN_SEND = 24
-
-// TelemetryCheckpoint - Checks if 24 hours has passed since telemetry was last sent. If so, sends telemetry data to posthog
-func TelemetryCheckpoint() error {
-
-	// if telemetry is turned off, return without doing anything
+// sendTelemetry - gathers telemetry data and sends to posthog
+func sendTelemetry() error {
 	if servercfg.Telemetry() == "off" {
 		return nil
 	}
-	// get the telemetry record in the DB, which contains a timestamp
-	telRecord, err := fetchTelemetryRecord()
+
+	var telRecord, err = fetchTelemetryRecord()
 	if err != nil {
 		return err
 	}
-	sendtime := time.Unix(telRecord.LastSend, 0).Add(time.Hour * time.Duration(TELEMETRY_HOURS_BETWEEN_SEND))
-	// can set to 2 minutes for testing
-	//sendtime := time.Unix(telRecord.LastSend, 0).Add(time.Minute * 2)
-	enoughTimeElapsed := time.Now().After(sendtime)
-	// if more than 24 hours has elapsed, send telemetry to posthog
-	if enoughTimeElapsed {
-		err = sendTelemetry(telRecord.UUID)
-		if err != nil {
-			logger.Log(1, err.Error())
-		}
-	}
-	return nil
-}
-
-// sendTelemetry - gathers telemetry data and sends to posthog
-func sendTelemetry(serverUUID string) error {
 	// get telemetry data
 	d, err := fetchTelemetryData()
 	if err != nil {
 		return err
 	}
-	client, err := posthog.NewWithConfig(POSTHOG_PUB_KEY, posthog.Config{Endpoint: POSTHOG_ENDPOINT})
+	client, err := posthog.NewWithConfig(posthog_pub_key, posthog.Config{Endpoint: posthog_endpoint})
 	if err != nil {
 		return err
 	}
 	defer client.Close()
 
 	// send to posthog
-	err = client.Enqueue(posthog.Capture{
-		DistinctId: serverUUID,
+	return client.Enqueue(posthog.Capture{
+		DistinctId: telRecord.UUID,
 		Event:      "daily checkin",
 		Properties: posthog.NewProperties().
 			Set("nodes", d.Nodes).
@@ -78,11 +55,6 @@ func sendTelemetry(serverUUID string) error {
 			Set("k8s", d.Count.K8S).
 			Set("version", d.Version),
 	})
-	if err != nil {
-		return err
-	}
-	//set telemetry timestamp for server, restarts 24 hour cycle
-	return setTelemetryTimestamp(serverUUID)
 }
 
 // fetchTelemetry - fetches telemetry data: count of various object types in DB
@@ -93,7 +65,7 @@ func fetchTelemetryData() (telemetryData, error) {
 	data.Users = getDBLength(database.USERS_TABLE_NAME)
 	data.Networks = getDBLength(database.NETWORKS_TABLE_NAME)
 	data.Version = servercfg.GetVersion()
-	nodes, err := logic.GetAllNodes()
+	nodes, err := GetAllNodes()
 	if err == nil {
 		data.Nodes = len(nodes)
 		data.Count = getClientCount(nodes)
@@ -108,7 +80,7 @@ func setTelemetryTimestamp(uuid string) error {
 		UUID:     uuid,
 		LastSend: lastsend,
 	}
-	jsonObj, err := json.Marshal(serverTelData)
+	jsonObj, err := json.Marshal(&serverTelData)
 	if err != nil {
 		return err
 	}

logic/timer.go

@@ -0,0 +1,62 @@
+package logic
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/gravitl/netmaker/logger"
+)
+
+// == Constants ==
+
+// How long to wait before sending telemetry to server (24 hours)
+const timer_hours_between_runs = 24
+
+// == Public ==
+
+// TimerCheckpoint - Checks if 24 hours has passed since telemetry was last sent. If so, sends telemetry data to posthog
+func TimerCheckpoint() error {
+	// get the telemetry record in the DB, which contains a timestamp
+	telRecord, err := fetchTelemetryRecord()
+	if err != nil {
+		return err
+	}
+	sendtime := time.Unix(telRecord.LastSend, 0).Add(time.Hour * time.Duration(timer_hours_between_runs))
+	// can set to 2 minutes for testing
+	// sendtime := time.Unix(telRecord.LastSend, 0).Add(time.Minute * 2)
+	enoughTimeElapsed := time.Now().After(sendtime)
+	// if more than 24 hours has elapsed, send telemetry to posthog
+	if enoughTimeElapsed {
+		// run any time hooks
+		runHooks()
+	}
+	// set telemetry timestamp for server, restarts 24 hour cycle
+	return setTelemetryTimestamp(telRecord.UUID)
+}
+
+// AddHook - adds a hook function to run every 24hrs
+func AddHook(ifaceToAdd interface{}) {
+	timeHooks = append(timeHooks, ifaceToAdd)
+}
+
+// == private ==
+
+// timeHooks - functions to run once a day, functions must take no parameters
+var timeHooks = []interface{}{
+	loggerDump,
+	sendTelemetry,
+}
+
+func loggerDump() error {
+	logger.DumpFile(fmt.Sprintf("data/netmaker.log.%s", time.Now().Format(logger.TimeFormatDay)))
+	return nil
+}
+
+// runHooks - runs the functions currently in the timeHooks data structure
+func runHooks() {
+	for _, hook := range timeHooks {
+		if err := hook.(func() error)(); err != nil {
+			logger.Log(1, "error occurred when running timer function:", err.Error())
+		}
+	}
+}

logic/wireguard.go

@@ -12,6 +12,7 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/models"
 	"github.com/gravitl/netmaker/netclient/ncutils"
+	"github.com/gravitl/netmaker/netclient/wireguard"
 	"golang.zx2c4.com/wireguard/wgctrl"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
@@ -84,17 +85,13 @@ func initWireguard(node *models.Node, privkey string, peers []wgtypes.PeerConfig
 	}
 
 	if !ncutils.IsKernel() {
-		var newConf string
-		newConf, _ = ncutils.CreateWireGuardConf(node, key.String(), strconv.FormatInt(int64(node.ListenPort), 10), peers)
-		confPath := ncutils.GetNetclientPathSpecific() + ifacename + ".conf"
-		logger.Log(1, "writing wg conf file to:", confPath)
-		err = os.WriteFile(confPath, []byte(newConf), 0644)
-		if err != nil {
-			logger.Log(1, "error writing wg conf file to", confPath, ":", err.Error())
+		if err := wireguard.WriteWgConfig(node, key.String(), peers); err != nil {
+			logger.Log(1, "error writing wg conf file: ", err.Error())
 			return err
 		}
 		// spin up userspace + apply the conf file
 		var deviceiface = ifacename
+		confPath := ncutils.GetNetclientPathSpecific() + ifacename + ".conf"
 		d, _ := wgclient.Device(deviceiface)
 		for d != nil && d.Name == deviceiface {
 			_ = RemoveConf(ifacename, false) // remove interface first

main.go

@@ -9,7 +9,9 @@ import (
 	"runtime/debug"
 	"strconv"
 	"sync"
+	"syscall"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/gravitl/netmaker/auth"
 	controller "github.com/gravitl/netmaker/controllers"
 	"github.com/gravitl/netmaker/database"
@@ -18,6 +20,7 @@ import (
 	"github.com/gravitl/netmaker/logger"
 	"github.com/gravitl/netmaker/logic"
 	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/mq"
 	"github.com/gravitl/netmaker/netclient/ncutils"
 	"github.com/gravitl/netmaker/servercfg"
 	"github.com/gravitl/netmaker/serverctl"
@@ -41,9 +44,9 @@ func initialize() { // Client Mode Prereq Check
 	}
 	logger.Log(0, "database successfully connected")
 
-	err = serverctl.TelemetryCheckpoint()
+	err = logic.TimerCheckpoint()
 	if err != nil {
-		logger.Log(1, "Failed to send telemetry: ", err.Error())
+		logger.Log(1, "Timer error occurred: ", err.Error())
 	}
 	var authProvider = auth.InitializeAuthProvider()
 	if authProvider != "" {
@@ -116,8 +119,14 @@ func startControllers() {
 		go controller.HandleRESTRequests(&waitnetwork)
 	}
 
-	if !servercfg.IsAgentBackend() && !servercfg.IsRestBackend() {
-		logger.Log(0, "No Server Mode selected, so nothing is being served! Set either Agent mode (AGENT_BACKEND) or Rest mode (REST_BACKEND) to 'true'.")
+	//Run MessageQueue
+	if servercfg.IsMessageQueueBackend() {
+		waitnetwork.Add(1)
+		go runMessageQueue(&waitnetwork)
+	}
+
+	if !servercfg.IsAgentBackend() && !servercfg.IsRestBackend() && !servercfg.IsMessageQueueBackend() {
+		logger.Log(0, "No Server Mode selected, so nothing is being served! Set Agent mode (AGENT_BACKEND) or Rest mode (REST_BACKEND) or MessageQueue (MESSAGEQUEUE_BACKEND) to 'true'.")
 	}
 
 	waitnetwork.Wait()
@@ -169,6 +178,41 @@ func runGRPC(wg *sync.WaitGroup) {
 	logger.Log(0, "Closed DB connection.")
 }
 
+// Should we be using a context vice a waitgroup????????????
+func runMessageQueue(wg *sync.WaitGroup) {
+	defer wg.Done()
+	//refactor netclient.functions.SetupMQTT so can be called from here
+	//setupMQTT
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(servercfg.GetMessageQueueEndpoint())
+	logger.Log(0, "setting broker "+servercfg.GetMessageQueueEndpoint())
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		logger.Log(0, "unable to connect to message queue broker, closing down")
+		return
+	}
+	//Set up Subscriptions
+	if servercfg.GetDebug() {
+		if token := client.Subscribe("#", 2, mq.DefaultHandler); token.Wait() && token.Error() != nil {
+			client.Disconnect(240)
+			logger.Log(0, "default subscription failed")
+		}
+	}
+	if token := client.Subscribe("ping/#", 2, mq.Ping); token.Wait() && token.Error() != nil {
+		client.Disconnect(240)
+		logger.Log(0, "ping subscription failed")
+	}
+	if token := client.Subscribe("update/#", 0, mq.UpdateNode); token.Wait() && token.Error() != nil {
+		client.Disconnect(240)
+		logger.Log(0, "node update subscription failed")
+	}
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
+	<-quit
+	logger.Log(0, "Message Queue shutting down")
+	client.Disconnect(250)
+}
+
 func authServerUnaryInterceptor() grpc.ServerOption {
 	return grpc.UnaryInterceptor(controller.AuthServerUnaryInterceptor)
 }

models/mqtt.go

@@ -0,0 +1,13 @@
+package models
+
+import "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+type PeerUpdate struct {
+	Network string
+	Peers   []wgtypes.PeerConfig
+}
+
+type KeyUpdate struct {
+	Network   string
+	Interface string
+}

mq/mq.go

@@ -0,0 +1,158 @@
+package mq
+
+import (
+	"encoding/json"
+	"errors"
+	"log"
+	"strings"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/gravitl/netmaker/database"
+	"github.com/gravitl/netmaker/logger"
+	"github.com/gravitl/netmaker/logic"
+	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/servercfg"
+)
+
+// DefaultHandler default message queue handler - only called when GetDebug == true
+var DefaultHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	logger.Log(0, "MQTT Message: Topic: ", string(msg.Topic()), " Message: ", string(msg.Payload()))
+}
+
+// Ping message Handler -- handles ping topic from client nodes
+var Ping mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	logger.Log(0, "Ping Handler: ", msg.Topic())
+	go func() {
+		id, err := GetID(msg.Topic())
+		if err != nil {
+			logger.Log(0, "error getting node.ID sent on ping topic ")
+			return
+		}
+		node, err := logic.GetNodeByID(id)
+		if err != nil {
+			logger.Log(0, "mq-ping error getting node: ", err.Error())
+			record, err := database.FetchRecord(database.NODES_TABLE_NAME, id)
+			if err != nil {
+				logger.Log(0, "error reading database ", err.Error())
+				return
+			}
+			logger.Log(0, "record from database")
+			logger.Log(0, record)
+			return
+		}
+		node.SetLastCheckIn()
+		if err := logic.UpdateNode(&node, &node); err != nil {
+			logger.Log(0, "error updating node ", err.Error())
+		}
+		logger.Log(0, "ping processed")
+		// --TODO --set client version once feature is implemented.
+		//node.SetClientVersion(msg.Payload())
+	}()
+}
+
+// UpdateNode  message Handler -- handles updates from client nodes
+var UpdateNode mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	go func() {
+		id, err := GetID(msg.Topic())
+		if err != nil {
+			logger.Log(1, "error getting node.ID sent on ", msg.Topic(), err.Error())
+			return
+		}
+		logger.Log(1, "Update Node Handler", id)
+		var newNode models.Node
+		if err := json.Unmarshal(msg.Payload(), &newNode); err != nil {
+			logger.Log(1, "error unmarshaling payload ", err.Error())
+			return
+		}
+		currentNode, err := logic.GetNodeByID(newNode.ID)
+		if err != nil {
+			logger.Log(1, "error getting node ", newNode.ID, err.Error())
+			return
+		}
+		if err := logic.UpdateNode(&currentNode, &newNode); err != nil {
+			logger.Log(1, "error saving node", err.Error())
+		}
+		if logic.ShouldPeersUpdate(&currentNode, &newNode) {
+			if err := PublishPeerUpdate(client, &newNode); err != nil {
+				logger.Log(1, "error publishing peer update ", err.Error())
+				return
+			}
+		}
+	}()
+}
+
+// PublishPeerUpdate --- deterines and publishes a peer update to all the peers of a node
+func PublishPeerUpdate(client mqtt.Client, newNode *models.Node) error {
+	networkNodes, err := logic.GetNetworkNodes(newNode.Network)
+	if err != nil {
+		logger.Log(1, "err getting Network Nodes", err.Error())
+		return err
+	}
+	for _, node := range networkNodes {
+		peerUpdate, err := logic.GetPeerUpdate(&node)
+		if err != nil {
+			logger.Log(1, "error getting peer update for node ", node.ID, err.Error())
+			continue
+		}
+		data, err := json.Marshal(&peerUpdate)
+		if err != nil {
+			logger.Log(2, "error marshaling peer update ", err.Error())
+			return err
+		}
+		if token := client.Publish("update/peers/"+node.ID, 0, false, data); token.Wait() && token.Error() != nil {
+			logger.Log(2, "error publishing peer update to peer ", node.ID, token.Error().Error())
+			return err
+		}
+	}
+	return nil
+}
+
+// GetID -- decodes a message queue topic and returns the embedded node.ID
+func GetID(topic string) (string, error) {
+	parts := strings.Split(topic, "/")
+	count := len(parts)
+	if count == 1 {
+		return "", errors.New("invalid topic")
+	}
+	//the last part of the topic will be the node.ID
+	return parts[count-1], nil
+}
+
+// UpdateNode -- publishes a node update
+func NodeUpdate(node *models.Node) error {
+	logger.Log(3, "publishing node update to "+node.Name)
+	client := SetupMQTT()
+	defer client.Disconnect(250)
+	data, err := json.Marshal(node)
+	if err != nil {
+		logger.Log(2, "error marshalling node update ", err.Error())
+		return err
+	}
+	if token := client.Publish("update/"+node.ID, 0, false, data); token.Wait() && token.Error() != nil {
+		logger.Log(2, "error publishing peer update to peer ", node.ID, token.Error().Error())
+		return err
+	}
+	return nil
+}
+
+// UpdatePeers -- publishes a peer update to all the peers of a node
+func UpdatePeers(node *models.Node) error {
+	client := SetupMQTT()
+	defer client.Disconnect(250)
+	if err := PublishPeerUpdate(client, node); err != nil {
+		return err
+	}
+	return nil
+}
+
+// SetupMQTT creates a connection to broker and return client
+func SetupMQTT() mqtt.Client {
+	opts := mqtt.NewClientOptions()
+	broker := servercfg.GetMessageQueueEndpoint()
+	opts.AddBroker(broker)
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	return client
+}

netclient/cli_options/cmds.go

@@ -118,5 +118,14 @@ func GetCommands(cliFlags []cli.Flag) []*cli.Command {
 				return err
 			},
 		},
+		{
+			Name:  "daemon",
+			Usage: "run netclient as daemon",
+			Flags: cliFlags,
+			Action: func(c *cli.Context) error {
+				err := command.Daemon()
+				return err
+			},
+		},
 	}
 }

netclient/command/commands.go

@@ -18,7 +18,7 @@ func Join(cfg config.ClientConfig, privateKey string) error {
 
 	var err error
 	err = functions.JoinNetwork(cfg, privateKey)
-	if err != nil && !cfg.DebugJoin {
+	if err != nil && !cfg.DebugOn {
 		if !strings.Contains(err.Error(), "ALREADY_INSTALLED") {
 			ncutils.PrintLog("error installing: "+err.Error(), 1)
 			err = functions.LeaveNetwork(cfg.Network)
@@ -210,3 +210,8 @@ func Uninstall() error {
 	ncutils.PrintLog("uninstalled netclient", 0)
 	return err
 }
+
+func Daemon() error {
+	err := functions.Daemon()
+	return err
+}

netclient/config/config.go

@@ -29,7 +29,7 @@ type ClientConfig struct {
 	Network         string         `yaml:"network"`
 	Daemon          string         `yaml:"daemon"`
 	OperatingSystem string         `yaml:"operatingsystem"`
-	DebugJoin       bool           `yaml:"debugjoin"`
+	DebugOn         bool           `yaml:"debugon"`
 }
 
 // ServerConfig - struct for dealing with the server information for a netclient

netclient/daemon/macos.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"path/filepath"
 
 	"github.com/gravitl/netmaker/netclient/ncutils"
 )
@@ -13,21 +14,22 @@ const MAC_SERVICE_NAME = "com.gravitl.netclient"
 // SetupMacDaemon - Creates a daemon service from the netclient under LaunchAgents for MacOS
 func SetupMacDaemon(interval string) error {
 
-	if !ncutils.FileExists("/etc/netclient/netclient") {
-		binarypath, err := os.Executable()
-		if err != nil {
-			return err
-		}
-		ncutils.PrintLog("installing binary from "+binarypath, 0)
-		err = ncutils.Copy(binarypath, "/etc/netclient/netclient")
+	dir, err := filepath.Abs(filepath.Dir(os.Args[0]))
+	if err != nil {
+		return err
+	}
+	binarypath := dir + "/netclient"
+
+	if !ncutils.FileExists(EXEC_DIR + "netclient") {
+		err = ncutils.Copy(binarypath, EXEC_DIR+"netclient")
 		if err != nil {
 			log.Println(err)
 			return err
 		}
 	}
 
-	_, err := os.Stat("~/Library/LaunchAgents")
-	if os.IsNotExist(err) {
+	_, errN := os.Stat("~/Library/LaunchAgents")
+	if os.IsNotExist(errN) {
 		os.Mkdir("~/Library/LaunchAgents", 0755)
 	}
 	err = CreateMacService(MAC_SERVICE_NAME, interval)
@@ -49,6 +51,7 @@ func CleanupMac() {
 	}
 
 	os.RemoveAll(ncutils.GetNetclientPath())
+	os.Remove(EXEC_DIR + "netclient")
 }
 
 // CreateMacService - Creates the mac service file for LaunchDaemons
@@ -78,10 +81,8 @@ func MacDaemonString(interval string) string {
 	<key>Label</key><string>com.gravitl.netclient</string>
 	<key>ProgramArguments</key>
 		<array>
-			<string>/etc/netclient/netclient</string>
-			<string>checkin</string>
-			<string>-n</string>
-			<string>all</string>
+			<string>/sbin/netclient</string>
+			<string>daemon</string>
 		</array>
 	<key>StandardOutPath</key><string>/etc/netclient/com.gravitl.netclient.log</string>
 	<key>StandardErrorPath</key><string>/etc/netclient/com.gravitl.netclient.log</string>

netclient/daemon/systemd.go

@@ -10,6 +10,8 @@ import (
 	"github.com/gravitl/netmaker/netclient/ncutils"
 )
 
+const EXEC_DIR = "/sbin/"
+
 // SetupSystemDDaemon - sets system daemon for supported machines
 func SetupSystemDDaemon(interval string) error {
 
@@ -29,12 +31,10 @@ func SetupSystemDDaemon(interval string) error {
 		log.Println("couldnt find or create /etc/netclient")
 		return err
 	}
-
-	if !ncutils.FileExists("/usr/local/bin/netclient") {
-		os.Symlink("/etc/netclient/netclient", "/usr/local/bin/netclient")
-	}
-	if !ncutils.FileExists("/etc/netclient/netclient") {
-		err = ncutils.Copy(binarypath, "/etc/netclient/netclient")
+	//install binary
+	//should check if the existing binary is the corect version -- for now only copy if file doesn't exist
+	if !ncutils.FileExists(EXEC_DIR + "netclient") {
+		err = ncutils.Copy(binarypath, EXEC_DIR+"netclient")
 		if err != nil {
 			log.Println(err)
 			return err
@@ -42,36 +42,17 @@ func SetupSystemDDaemon(interval string) error {
 	}
 
 	systemservice := `[Unit]
-Description=Network Check
-Wants=netclient.timer
+Description=Netclient message queue
 
 [Service]
 Type=simple
-ExecStart=/etc/netclient/netclient checkin -n all
+ExecStart=/usr/sbin/netclient daemon
 
 [Install]
 WantedBy=multi-user.target
-`
-
-	systemtimer := `[Unit]
-Description=Calls the Netmaker Mesh Client Service
-Requires=netclient.service
-
-[Timer]
-Unit=netclient.service
-
-`
-	systemtimer = systemtimer + "OnCalendar=*:*:0/" + interval
-
-	systemtimer = systemtimer +
-		`
-
-[Install]
-WantedBy=timers.target
 `
 
 	servicebytes := []byte(systemservice)
-	timerbytes := []byte(systemtimer)
 
 	if !ncutils.FileExists("/etc/systemd/system/netclient.service") {
 		err = os.WriteFile("/etc/systemd/system/netclient.service", servicebytes, 0644)
@@ -80,25 +61,17 @@ WantedBy=timers.target
 			return err
 		}
 	}
-
-	if !ncutils.FileExists("/etc/systemd/system/netclient.timer") {
-		err = os.WriteFile("/etc/systemd/system/netclient.timer", timerbytes, 0644)
-		if err != nil {
-			log.Println(err)
-			return err
-		}
-	}
-
 	_, _ = ncutils.RunCmd("systemctl enable netclient.service", true)
 	_, _ = ncutils.RunCmd("systemctl daemon-reload", true)
-	_, _ = ncutils.RunCmd("systemctl enable netclient.timer", true)
-	_, _ = ncutils.RunCmd("systemctl start netclient.timer", true)
+	_, _ = ncutils.RunCmd("systemctl start netclient.service", true)
 	return nil
 }
 
 func CleanupLinux() {
-	err := os.RemoveAll(ncutils.GetNetclientPath())
-	if err != nil {
+	if err := os.RemoveAll(ncutils.GetNetclientPath()); err != nil {
+		ncutils.PrintLog("Removing netclient configs: "+err.Error(), 1)
+	}
+	if err := os.Remove(EXEC_DIR + "netclient"); err != nil {
 		ncutils.PrintLog("Removing netclient binary: "+err.Error(), 1)
 	}
 }

netclient/functions/daemon.go

@@ -0,0 +1,293 @@
+package functions
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"os"
+	"os/signal"
+	"runtime"
+	"syscall"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/netclient/config"
+	"github.com/gravitl/netmaker/netclient/local"
+	"github.com/gravitl/netmaker/netclient/ncutils"
+	"github.com/gravitl/netmaker/netclient/wireguard"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+// Daemon runs netclient daemon from command line
+func Daemon() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	networks, err := ncutils.GetSystemNetworks()
+	if err != nil {
+		cancel()
+		return err
+	}
+	for _, network := range networks {
+		go MessageQueue(ctx, network)
+	}
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
+	<-quit
+	cancel()
+	ncutils.Log("all done")
+	return nil
+}
+
+// SetupMQTT creates a connection to broker and return client
+func SetupMQTT(cfg *config.ClientConfig) mqtt.Client {
+	opts := mqtt.NewClientOptions()
+	ncutils.Log("setting broker to " + cfg.Server.CoreDNSAddr + ":1883")
+	opts.AddBroker(cfg.Server.CoreDNSAddr + ":1883")
+	opts.SetDefaultPublishHandler(All)
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	return client
+}
+
+// MessageQueue sets up Message Queue and subsribes/publishes updates to/from server
+func MessageQueue(ctx context.Context, network string) {
+	ncutils.Log("netclient go routine started for " + network)
+	var cfg config.ClientConfig
+	cfg.Network = network
+	cfg.ReadConfig()
+	ncutils.Log("daemon started for network:" + network)
+	client := SetupMQTT(&cfg)
+	if cfg.DebugOn {
+		if token := client.Subscribe("#", 0, nil); token.Wait() && token.Error() != nil {
+			log.Fatal(token.Error())
+		}
+		ncutils.Log("subscribed to all topics for debugging purposes")
+	}
+	if token := client.Subscribe("update/"+cfg.Node.ID, 0, NodeUpdate); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	if cfg.DebugOn {
+		ncutils.Log("subscribed to node updates for node " + cfg.Node.Name + " update/" + cfg.Node.ID)
+	}
+	if token := client.Subscribe("update/peers/"+cfg.Node.ID, 0, UpdatePeers); token.Wait() && token.Error() != nil {
+		log.Fatal(token.Error())
+	}
+	if cfg.DebugOn {
+		ncutils.Log("subscribed to node updates for node " + cfg.Node.Name + " update/peers/" + cfg.Node.ID)
+	}
+	defer client.Disconnect(250)
+	go Checkin(ctx, &cfg, network)
+	<-ctx.Done()
+	ncutils.Log("shutting down daemon")
+}
+
+// All -- mqtt message hander for all ('#') topics
+var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("default message handler -- received message but not handling")
+	ncutils.Log("Topic: " + string(msg.Topic()))
+	//ncutils.Log("Message: " + string(msg.Payload()))
+}
+
+// NodeUpdate -- mqtt message handler for /update/<NodeID> topic
+var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	ncutils.Log("received message to update node " + string(msg.Payload()))
+	//potentiall blocking i/o so do this in a go routine
+	go func() {
+		var newNode models.Node
+		var cfg config.ClientConfig
+		err := json.Unmarshal(msg.Payload(), &newNode)
+		if err != nil {
+			ncutils.Log("error unmarshalling node update data" + err.Error())
+			return
+		}
+		cfg.Network = newNode.Network
+		cfg.ReadConfig()
+		//check if interface name has changed if so delete.
+		if cfg.Node.Interface != newNode.Interface {
+			if err = wireguard.RemoveConf(cfg.Node.Interface, true); err != nil {
+				ncutils.PrintLog("could not delete old interface "+cfg.Node.Interface+": "+err.Error(), 1)
+			}
+		}
+		newNode.PullChanges = "no"
+		//ensure that OS never changes
+		newNode.OS = runtime.GOOS
+		cfg.Node = newNode
+		switch newNode.Action {
+		case models.NODE_DELETE:
+			if err := RemoveLocalInstance(&cfg, cfg.Network); err != nil {
+				ncutils.PrintLog("error deleting local instance: "+err.Error(), 1)
+				return
+			}
+			if token := client.Unsubscribe("update/"+newNode.ID, "update/peers/"+newNode.ID); token.Wait() && token.Error() != nil {
+				ncutils.PrintLog("error unsubscribing during node deletion", 1)
+			}
+			return
+		case models.NODE_UPDATE_KEY:
+			if err := UpdateKeys(&cfg, client); err != nil {
+				ncutils.PrintLog("err updating wireguard keys: "+err.Error(), 1)
+			}
+		case models.NODE_NOOP:
+		default:
+		}
+		//Save new config
+		if err := config.Write(&cfg, cfg.Network); err != nil {
+			ncutils.PrintLog("error updating node configuration: "+err.Error(), 1)
+		}
+		nameserver := cfg.Server.CoreDNSAddr
+		privateKey, err := wireguard.RetrievePrivKey(newNode.Network)
+		if err != nil {
+			ncutils.Log("error reading PrivateKey " + err.Error())
+			return
+		}
+		file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
+		if err := wireguard.UpdateWgInterface(file, privateKey, nameserver, newNode); err != nil {
+			ncutils.Log("error updating wireguard config " + err.Error())
+			return
+		}
+		ncutils.Log("applyWGQuickConf to " + file)
+		err = wireguard.ApplyWGQuickConf(file)
+		if err != nil {
+			ncutils.Log("error restarting wg after node update " + err.Error())
+			return
+		}
+		//deal with DNS
+		if newNode.DNSOn == "yes" {
+			ncutils.Log("setting up DNS")
+			if err = local.UpdateDNS(cfg.Node.Interface, cfg.Network, cfg.Server.CoreDNSAddr); err != nil {
+				ncutils.Log("error applying dns" + err.Error())
+			}
+		} else {
+			ncutils.Log("settng DNS off")
+			_, err := ncutils.RunCmd("/usr/bin/resolvectl revert "+cfg.Node.Interface, true)
+			if err != nil {
+				ncutils.Log("error applying dns" + err.Error())
+			}
+		}
+	}()
+}
+
+// UpdatePeers -- mqtt message handler for /update/peers/<NodeID> topic
+var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+	go func() {
+		var peerUpdate models.PeerUpdate
+		err := json.Unmarshal(msg.Payload(), &peerUpdate)
+		if err != nil {
+			ncutils.Log("error unmarshalling peer data")
+			return
+		}
+		ncutils.Log("update peer handler")
+		var cfg config.ClientConfig
+		cfg.Network = peerUpdate.Network
+		cfg.ReadConfig()
+		file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
+		err = wireguard.UpdateWgPeers(file, peerUpdate.Peers)
+		if err != nil {
+			ncutils.Log("error updating wireguard peers" + err.Error())
+			return
+		}
+		ncutils.Log("applyWGQuickConf to " + file)
+		err = wireguard.ApplyWGQuickConf(file)
+		if err != nil {
+			ncutils.Log("error restarting wg after peer update " + err.Error())
+			return
+		}
+	}()
+}
+
+// UpdateKeys -- updates private key and returns new publickey
+func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) error {
+	ncutils.Log("received message to update keys")
+	//potentiall blocking i/o so do this in a go routine
+	key, err := wgtypes.GeneratePrivateKey()
+	if err != nil {
+		ncutils.Log("error generating privatekey " + err.Error())
+		return err
+	}
+	file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
+	if err := wireguard.UpdatePrivateKey(file, key.String()); err != nil {
+		ncutils.Log("error updating wireguard key " + err.Error())
+		return err
+	}
+	cfg.Node.PublicKey = key.PublicKey().String()
+	PublishNodeUpdate(cfg)
+	if err := config.ModConfig(&cfg.Node); err != nil {
+		ncutils.Log("error updating local config " + err.Error())
+	}
+	return nil
+}
+
+// Checkin  -- go routine that checks for public or local ip changes, publishes changes
+//   if there are no updates, simply "pings" the server as a checkin
+func Checkin(ctx context.Context, cfg *config.ClientConfig, network string) {
+	for {
+		select {
+		case <-ctx.Done():
+			ncutils.Log("Checkin cancelled")
+			return
+			//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
+		case <-time.After(time.Second * 60):
+			ncutils.Log("Checkin running")
+			//read latest config
+			cfg.ReadConfig()
+			if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
+				extIP, err := ncutils.GetPublicIP()
+				if err != nil {
+					ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
+				}
+				if cfg.Node.Endpoint != extIP && extIP != "" {
+					ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+extIP, 1)
+					cfg.Node.Endpoint = extIP
+					PublishNodeUpdate(cfg)
+				}
+				intIP, err := getPrivateAddr()
+				if err != nil {
+					ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
+				}
+				if cfg.Node.LocalAddress != intIP && intIP != "" {
+					ncutils.PrintLog("local Address has changed from "+cfg.Node.LocalAddress+" to "+intIP, 1)
+					cfg.Node.LocalAddress = intIP
+					PublishNodeUpdate(cfg)
+				}
+			} else {
+				localIP, err := ncutils.GetLocalIP(cfg.Node.LocalRange)
+				if err != nil {
+					ncutils.PrintLog("error encountered checking ip addresses: "+err.Error(), 1)
+				}
+				if cfg.Node.Endpoint != localIP && localIP != "" {
+					ncutils.PrintLog("endpoint has changed from "+cfg.Node.Endpoint+" to "+localIP, 1)
+					cfg.Node.Endpoint = localIP
+					PublishNodeUpdate(cfg)
+				}
+			}
+			Hello(cfg, network)
+			ncutils.Log("Checkin complete")
+		}
+	}
+}
+
+// PublishNodeUpdates -- saves node and pushes changes to broker
+func PublishNodeUpdate(cfg *config.ClientConfig) {
+	if err := config.Write(cfg, cfg.Network); err != nil {
+		ncutils.Log("error saving configuration" + err.Error())
+	}
+	client := SetupMQTT(cfg)
+	data, err := json.Marshal(cfg.Node)
+	if err != nil {
+		ncutils.Log("error marshling node update " + err.Error())
+	}
+	if token := client.Publish("update/"+cfg.Node.ID, 0, false, data); token.Wait() && token.Error() != nil {
+		ncutils.Log("error publishing endpoint update " + token.Error().Error())
+	}
+	client.Disconnect(250)
+}
+
+// Hello -- ping the broker to let server know node is alive and doing fine
+func Hello(cfg *config.ClientConfig, network string) {
+	client := SetupMQTT(cfg)
+	if token := client.Publish("ping/"+cfg.Node.ID, 2, false, "hello world!"); token.Wait() && token.Error() != nil {
+		ncutils.Log("error publishing ping " + token.Error().Error())
+	}
+	client.Disconnect(250)
+}

netclient/ncutils/netclientutils.go

@@ -164,7 +164,7 @@ func GetMacAddr() ([]string, error) {
 func parsePeers(keepalive int32, peers []wgtypes.PeerConfig) (string, error) {
 	peersString := ""
 	if keepalive <= 0 {
-		keepalive = 20
+		keepalive = 0
 	}
 
 	for _, peer := range peers {

netclient/ncutils/netclientutils_darwin.go

@@ -1,14 +1,9 @@
 package ncutils
 
 import (
-	"fmt"
 	"log"
 	"os/exec"
-	"strconv"
 	"strings"
-
-	"github.com/gravitl/netmaker/models"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
 // RunCmd - runs a local command
@@ -34,32 +29,4 @@ func GetEmbedded() error {
 	return nil
 }
 
-// CreateWireGuardConf - creates a WireGuard conf string
-func CreateWireGuardConf(node *models.Node, privatekey string, listenPort string, peers []wgtypes.PeerConfig) (string, error) {
-	peersString, err := parsePeers(node.PersistentKeepalive, peers)
-	var listenPortString string
-	if node.MTU <= 0 {
-		node.MTU = 1280
-	}
-	if listenPort != "" {
-		listenPortString += "ListenPort = " + listenPort
-	}
-	if err != nil {
-		return "", err
-	}
-	config := fmt.Sprintf(`[Interface]
-Address = %s
-PrivateKey = %s
-MTU = %s
-%s
-
-%s
 
-`,
-		node.Address+"/32",
-		privatekey,
-		strconv.Itoa(int(node.MTU)),
-		listenPortString,
-		peersString)
-	return config, nil
-}

netclient/ncutils/netclientutils_freebsd.go

@@ -2,16 +2,11 @@ package ncutils
 
 import (
 	"context"
-	"fmt"
 	"log"
 	"os/exec"
-	"strconv"
 	"strings"
 	"syscall"
 	"time"
-
-	"github.com/gravitl/netmaker/models"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
 // RunCmdFormatted - run a command formatted for freebsd
@@ -43,32 +38,4 @@ func RunCmd(command string, printerr bool) (string, error) {
 	return string(out), err
 }
 
-// CreateWireGuardConf - creates a WireGuard conf string
-func CreateWireGuardConf(node *models.Node, privatekey string, listenPort string, peers []wgtypes.PeerConfig) (string, error) {
-	peersString, err := parsePeers(node.PersistentKeepalive, peers)
-	var listenPortString string
-	if node.MTU <= 0 {
-		node.MTU = 1280
-	}
-	if listenPort != "" {
-		listenPortString += "ListenPort = " + listenPort
-	}
-	if err != nil {
-		return "", err
-	}
-	config := fmt.Sprintf(`[Interface]
-Address = %s
-PrivateKey = %s
-MTU = %s
-%s
-
-%s
 
-`,
-		node.Address+"/32",
-		privatekey,
-		strconv.Itoa(int(node.MTU)),
-		listenPortString,
-		peersString)
-	return config, nil
-}

netclient/ncutils/netclientutils_linux.go

@@ -3,11 +3,7 @@ package ncutils
 import (
 	"fmt"
 	"os/exec"
-	"strconv"
 	"strings"
-
-	"github.com/gravitl/netmaker/models"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
 // RunCmd - runs a local command
@@ -33,44 +29,4 @@ func GetEmbedded() error {
 	return nil
 }
 
-// CreateWireGuardConf - creates a user space WireGuard conf
-func CreateWireGuardConf(node *models.Node, privatekey string, listenPort string, peers []wgtypes.PeerConfig) (string, error) {
-	peersString, err := parsePeers(node.PersistentKeepalive, peers)
-	var listenPortString, postDownString, postUpString string
-	if node.MTU <= 0 {
-		node.MTU = 1280
-	}
-	if node.PostDown != "" {
-		postDownString = fmt.Sprintf("PostDown = %s", node.PostDown)
-	}
-	if node.PostUp != "" {
-		postUpString = fmt.Sprintf("PostUp = %s", node.PostUp)
-	}
-
-	if listenPort != "" {
-		listenPortString = fmt.Sprintf("ListenPort = %s", listenPort)
-	}
-
-	if err != nil {
-		return "", err
-	}
-	config := fmt.Sprintf(`[Interface]
-Address = %s
-PrivateKey = %s
-MTU = %s
-%s
-%s
-%s
 
-%s
-
-`,
-		node.Address+"/32",
-		privatekey,
-		strconv.Itoa(int(node.MTU)),
-		postDownString,
-		postUpString,
-		listenPortString,
-		peersString)
-	return config, nil
-}

netclient/ncutils/netclientutils_windows.go

@@ -6,12 +6,8 @@ import (
 	"log"
 	"os"
 	"os/exec"
-	"strconv"
 	"strings"
 	"syscall"
-
-	"github.com/gravitl/netmaker/models"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
 //go:embed windowsdaemon/winsw.exe
@@ -48,36 +44,6 @@ func RunCmdFormatted(command string, printerr bool) (string, error) {
 	return string(out), err
 }
 
-// CreateWireGuardConf - creates a WireGuard conf string
-func CreateWireGuardConf(node *models.Node, privatekey string, listenPort string, peers []wgtypes.PeerConfig) (string, error) {
-	peersString, err := parsePeers(node.PersistentKeepalive, peers)
-	var listenPortString string
-	if node.MTU <= 0 {
-		node.MTU = 1280
-	}
-	if listenPort != "" {
-		listenPortString += "ListenPort = " + listenPort
-	}
-	if err != nil {
-		return "", err
-	}
-	config := fmt.Sprintf(`[Interface]
-Address = %s
-PrivateKey = %s
-MTU = %s
-%s
-
-%s
-
-`,
-		node.Address+"/32",
-		privatekey,
-		strconv.Itoa(int(node.MTU)),
-		listenPortString,
-		peersString)
-	return config, nil
-}
-
 // GetEmbedded - Gets the Windows daemon creator
 func GetEmbedded() error {
 	data, err := winswContent.ReadFile("windowsdaemon/winsw.exe")

netclient/wireguard/common.go

@@ -3,7 +3,6 @@ package wireguard
 import (
 	"errors"
 	"log"
-	"os"
 	"runtime"
 	"strconv"
 	"strings"
@@ -16,6 +15,12 @@ import (
 	"github.com/gravitl/netmaker/netclient/server"
 	"golang.zx2c4.com/wireguard/wgctrl"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"gopkg.in/ini.v1"
+)
+
+const (
+	section_interface = "Interface"
+	section_peers     = "Peer"
 )
 
 // SetPeers - sets peers on a given WireGuard interface
@@ -138,29 +143,16 @@ func InitWireguard(node *models.Node, privkey string, peers []wgtypes.PeerConfig
 	if node.Address == "" {
 		log.Fatal("no address to configure")
 	}
-	var newConf string
 	if node.UDPHolePunch != "yes" {
-		newConf, _ = ncutils.CreateWireGuardConf(node, key.String(), strconv.FormatInt(int64(node.ListenPort), 10), peers)
-	} else {
-		newConf, _ = ncutils.CreateWireGuardConf(node, key.String(), "", peers)
+		node.ListenPort = 0
 	}
-	confPath := ncutils.GetNetclientPathSpecific() + ifacename + ".conf"
-	ncutils.PrintLog("writing wg conf file to: "+confPath, 1)
-	err = os.WriteFile(confPath, []byte(newConf), 0644)
-	if err != nil {
-		ncutils.PrintLog("error writing wg conf file to "+confPath+": "+err.Error(), 1)
+	if err := WriteWgConfig(&modcfg.Node, key.String(), peers); err != nil {
+		ncutils.PrintLog("error writing wg conf file: "+err.Error(), 1)
 		return err
 	}
-	if ncutils.IsWindows() {
-		wgConfPath := ncutils.GetWGPathSpecific() + ifacename + ".conf"
-		err = os.WriteFile(wgConfPath, []byte(newConf), 0644)
-		if err != nil {
-			ncutils.PrintLog("error writing wg conf file to "+wgConfPath+": "+err.Error(), 1)
-			return err
-		}
-		confPath = wgConfPath
-	}
+
 	// spin up userspace / windows interface + apply the conf file
+	confPath := ncutils.GetNetclientPathSpecific() + ifacename + ".conf"
 	var deviceiface string
 	if ncutils.IsMac() {
 		deviceiface, err = local.GetMacIface(node.Address)
@@ -289,3 +281,159 @@ func ApplyConf(node models.Node, ifacename string, confPath string) error {
 	}
 	return err
 }
+
+// WriteWgConfig - creates a wireguard config file
+//func WriteWgConfig(cfg *config.ClientConfig, privateKey string, peers []wgtypes.PeerConfig) error {
+func WriteWgConfig(node *models.Node, privateKey string, peers []wgtypes.PeerConfig) error {
+	options := ini.LoadOptions{
+		AllowNonUniqueSections: true,
+		AllowShadows:           true,
+	}
+	wireguard := ini.Empty(options)
+	wireguard.Section(section_interface).Key("PrivateKey").SetValue(privateKey)
+	if node.ListenPort > 0 {
+		wireguard.Section(section_interface).Key("ListenPort").SetValue(strconv.Itoa(int(node.ListenPort)))
+	}
+	if node.Address != "" {
+		wireguard.Section(section_interface).Key("Address").SetValue(node.Address)
+	}
+	if node.Address6 != "" {
+		wireguard.Section(section_interface).Key("Address").SetValue(node.Address6)
+	}
+	// need to figure out DNS
+	//if node.DNSOn == "yes" {
+	//	wireguard.Section(section_interface).Key("DNS").SetValue(cfg.Server.CoreDNSAddr)
+	//}
+	if node.PostUp != "" {
+		wireguard.Section(section_interface).Key("PostUp").SetValue(node.PostUp)
+	}
+	if node.PostDown != "" {
+		wireguard.Section(section_interface).Key("PostDown").SetValue(node.PostDown)
+	}
+	if node.MTU != 0 {
+		wireguard.Section(section_interface).Key("MTU").SetValue(strconv.FormatInt(int64(node.MTU), 10))
+	}
+	for i, peer := range peers {
+		wireguard.SectionWithIndex(section_peers, i).Key("PublicKey").SetValue(peer.PublicKey.String())
+		if peer.PresharedKey != nil {
+			wireguard.SectionWithIndex(section_peers, i).Key("PreSharedKey").SetValue(peer.PresharedKey.String())
+		}
+		if peer.AllowedIPs != nil {
+			var allowedIPs string
+			for i, ip := range peer.AllowedIPs {
+				if i == 0 {
+					allowedIPs = ip.String()
+				} else {
+					allowedIPs = allowedIPs + ", " + ip.String()
+				}
+			}
+			wireguard.SectionWithIndex(section_peers, i).Key("AllowedIps").SetValue(allowedIPs)
+		}
+		if peer.Endpoint != nil {
+			wireguard.SectionWithIndex(section_peers, i).Key("Endpoint").SetValue(peer.Endpoint.String())
+		}
+
+		if peer.PersistentKeepaliveInterval != nil && peer.PersistentKeepaliveInterval.Seconds() > 0 {
+			wireguard.SectionWithIndex(section_peers, i).Key("PersistentKeepalive").SetValue(strconv.FormatInt((int64)(peer.PersistentKeepaliveInterval.Seconds()), 10))
+		}
+	}
+	if err := wireguard.SaveTo(ncutils.GetNetclientPathSpecific() + node.Interface + ".conf"); err != nil {
+		return err
+	}
+	return nil
+}
+
+// UpdateWgPeers - updates the peers of a network
+func UpdateWgPeers(file string, peers []wgtypes.PeerConfig) error {
+	options := ini.LoadOptions{
+		AllowNonUniqueSections: true,
+		AllowShadows:           true,
+	}
+	ncutils.Log("updating " + file)
+	wireguard, err := ini.LoadSources(options, file)
+	if err != nil {
+		return err
+	}
+	//delete the peers sections as they are going to be replaced
+	wireguard.DeleteSection(section_peers)
+	for i, peer := range peers {
+		wireguard.SectionWithIndex(section_peers, i).Key("PublicKey").SetValue(peer.PublicKey.String())
+		if peer.PresharedKey != nil {
+			wireguard.SectionWithIndex(section_peers, i).Key("PreSharedKey").SetValue(peer.PresharedKey.String())
+		}
+		if peer.AllowedIPs != nil {
+			var allowedIPs string
+			for i, ip := range peer.AllowedIPs {
+				if i == 0 {
+					allowedIPs = ip.String()
+				} else {
+					allowedIPs = allowedIPs + ", " + ip.String()
+				}
+			}
+			wireguard.SectionWithIndex(section_peers, i).Key("AllowedIps").SetValue(allowedIPs)
+		}
+		if peer.Endpoint != nil {
+			wireguard.SectionWithIndex(section_peers, i).Key("Endpoint").SetValue(peer.Endpoint.String())
+		}
+		if peer.PersistentKeepaliveInterval != nil && peer.PersistentKeepaliveInterval.Seconds() > 0 {
+			wireguard.SectionWithIndex(section_peers, i).Key("PersistentKeepalive").SetValue(strconv.FormatInt((int64)(peer.PersistentKeepaliveInterval.Seconds()), 10))
+		}
+	}
+	if err := wireguard.SaveTo(file); err != nil {
+		return err
+	}
+	return nil
+}
+
+// UpdateWgInterface - updates the interface section of a wireguard config file
+func UpdateWgInterface(file, privateKey, nameserver string, node models.Node) error {
+	options := ini.LoadOptions{
+		AllowNonUniqueSections: true,
+		AllowShadows:           true,
+	}
+	wireguard, err := ini.LoadSources(options, file)
+	if err != nil {
+		return err
+	}
+	wireguard.Section(section_interface).Key("PrivateKey").SetValue(privateKey)
+	wireguard.Section(section_interface).Key("ListenPort").SetValue(strconv.Itoa(int(node.ListenPort)))
+	if node.Address != "" {
+		wireguard.Section(section_interface).Key("Address").SetValue(node.Address)
+	}
+	if node.Address6 != "" {
+		wireguard.Section(section_interface).Key("Address").SetValue(node.Address6)
+	}
+	//if node.DNSOn == "yes" {
+	//	wireguard.Section(section_interface).Key("DNS").SetValue(nameserver)
+	//}
+	if node.PostUp != "" {
+		wireguard.Section(section_interface).Key("PostUp").SetValue(node.PostUp)
+	}
+	if node.PostDown != "" {
+		wireguard.Section(section_interface).Key("PostDown").SetValue(node.PostDown)
+	}
+	if node.MTU != 0 {
+		wireguard.Section(section_interface).Key("MTU").SetValue(strconv.FormatInt(int64(node.MTU), 10))
+	}
+	if err := wireguard.SaveTo(file); err != nil {
+		return err
+	}
+	return nil
+}
+
+// UpdatePrivateKey - updates the private key of a wireguard config file
+func UpdatePrivateKey(file, privateKey string) error {
+	options := ini.LoadOptions{
+		AllowNonUniqueSections: true,
+		AllowShadows:           true,
+	}
+	wireguard, err := ini.LoadSources(options, file)
+	if err != nil {
+		return err
+	}
+	wireguard.Section(section_interface).Key("PrivateKey").SetValue(privateKey)
+	if err := wireguard.SaveTo(file); err != nil {
+		return err
+	}
+	return nil
+}

netclient/wireguard/unix.go

@@ -53,8 +53,19 @@ func SetWGKeyConfig(network string, serveraddr string) error {
 
 // ApplyWGQuickConf - applies wg-quick commands if os supports
 func ApplyWGQuickConf(confPath string) error {
-	_, _ = ncutils.RunCmd("wg-quick down "+confPath, false)
-	_, err := ncutils.RunCmd("wg-quick up "+confPath, false)
+	_, err := os.Stat(confPath)
+	if err != nil {
+		ncutils.Log(confPath + " does not exist " + err.Error())
+		return err
+	}
+	_, err = ncutils.RunCmd("wg-quick down "+confPath, true)
+	if err != nil {
+		ncutils.Log("err runing wg-quick down " + confPath + err.Error())
+	}
+	_, err = ncutils.RunCmd("wg-quick up "+confPath, true)
+	if err != nil {
+		ncutils.Log("err runing wg-quick up " + confPath + err.Error())
+	}
 	return err
 }
 

scripts/install-netmaker.sh

@@ -136,6 +136,9 @@ echo "Setting Caddyfile..."
 sed -i "s/NETMAKER_BASE_DOMAIN/$NETMAKER_BASE_DOMAIN/g" /root/Caddyfile
 sed -i "s/YOUR_EMAIL/$EMAIL/g" /root/Caddyfile
 
+echo "Setting Broker..."
+sed -i "s/SERVER_PBLIC_IP/$SERVER_PUBLIC_IP/g" /root/mosquitto.conf
+
 echo "Setting docker-compose..."
 
 sed -i "s/NETMAKER_BASE_DOMAIN/$NETMAKER_BASE_DOMAIN/g" /root/docker-compose.yml

scripts/nm-quick.sh

@@ -135,6 +135,10 @@ wget -q -O /root/Caddyfile https://raw.githubusercontent.com/gravitl/netmaker/ma
 sed -i "s/NETMAKER_BASE_DOMAIN/$NETMAKER_BASE_DOMAIN/g" /root/Caddyfile
 sed -i "s/YOUR_EMAIL/$EMAIL/g" /root/Caddyfile
 
+echo "setting mosquitto.conf..."
+
+wget -q -O /root/mosquitto.conf https://raw.githubusercontent.com/gravil/netmaker/master/docker/mosquitto.conf
+sed -i "s/SERVER_PUBLIC_IP/$SERVER_PUBLIC_IP/g" /root/mosquitto.conf
 
 echo "setting docker-compose..."
 

servercfg/serverconf.go

@@ -85,6 +85,7 @@ func GetServerConfig() config.ServerConfig {
 	} else {
 		cfg.RCE = "off"
 	}
+	cfg.Debug = GetDebug()
 	cfg.Telemetry = Telemetry()
 	cfg.ManageIPTables = ManageIPTables()
 	services := strings.Join(GetPortForwardServiceList(), ",")
@@ -248,6 +249,18 @@ func GetGRPCPort() string {
 	return grpcport
 }
 
+// GetMessageQueueEndpoint - gets the message queue endpoint
+func GetMessageQueueEndpoint() string {
+	host, _ := GetPublicIP()
+	if os.Getenv("MQ_HOST") != "" {
+		host = os.Getenv("MQ_HOST")
+	} else if config.Config.Server.MQHOST != "" {
+		host = config.Config.Server.MQHOST
+	}
+	//Do we want MQ port configurable???
+	return host + ":1883"
+}
+
 // GetMasterKey - gets the configured master key of server
 func GetMasterKey() string {
 	key := "secretkey"
@@ -311,6 +324,21 @@ func IsAgentBackend() bool {
 	return isagent
 }
 
+// IsMessageQueueBackend - checks if message queue is on or off
+func IsMessageQueueBackend() bool {
+	ismessagequeue := true
+	if os.Getenv("MESSAGEQUEUE_BACKEND") != "" {
+		if os.Getenv("MESSAGEQUEUE_BACKEND") == "off" {
+			ismessagequeue = false
+		}
+	} else if config.Config.Server.MessageQueueBackend != "" {
+		if config.Config.Server.MessageQueueBackend == "off" {
+			ismessagequeue = false
+		}
+	}
+	return ismessagequeue
+}
+
 // IsClientMode - checks if it should run in client mode
 func IsClientMode() string {
 	isclient := "on"
@@ -581,3 +609,8 @@ func getMacAddr() string {
 func GetRce() bool {
 	return os.Getenv("RCE") == "on" || config.Config.Server.RCE == "on"
 }
+
+// GetDebug -- checks if debugging is enabled, off by default
+func GetDebug() bool {
+	return os.Getenv("DEBUG") == "on" || config.Config.Server.Debug == true
+}

serverctl/serverq.go

@@ -1,29 +0,0 @@
-package serverctl
-
-import (
-	"fmt"
-
-	"github.com/gravitl/netmaker/models"
-)
-
-// ServerQueue - holds data to be updated across the server
-var ServerQueue chan models.ServerUpdateData
-
-func init() {
-	ServerQueue = make(chan models.ServerUpdateData, 100)
-}
-
-// Push - Pushes ServerUpdateData to be used later
-func Push(serverData models.ServerUpdateData) {
-	ServerQueue <- serverData
-}
-
-// Pop - fetches first available data from queue
-func Pop() (models.ServerUpdateData, error) {
-	select {
-	case serverData := <-ServerQueue:
-		return serverData, nil
-	default:
-		return models.ServerUpdateData{}, fmt.Errorf("empty server queue")
-	}
-}