WIP commit to enable rebasing to uuid feature

main.go

@@ -9,6 +9,7 @@ import (
 	"runtime/debug"
 	"strconv"
 	"sync"
+	"syscall"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/gravitl/netmaker/auth"
@@ -117,9 +118,25 @@ func startControllers() {
 		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'.")
 	}
 
+	//if servercfg.IsClientMode() == "on" {
+	//	var checkintime = time.Duration(servercfg.GetServerCheckinInterval()) * time.Second
+	//	for { // best effort currently
+	//		var serverGroup sync.WaitGroup
+	//		serverGroup.Add(1)
+	//		go runClient(&serverGroup)
+	//		serverGroup.Wait()
+	//		time.Sleep(checkintime)
+	//	}
+	//}
+
 	waitnetwork.Wait()
 }
 
+func runClient(wg *sync.WaitGroup) {
+	defer wg.Done()
+	go serverctl.HandleContainedClient()
+}
+
 func runGRPC(wg *sync.WaitGroup) {
 
 	defer wg.Done()
@@ -187,7 +204,7 @@ func runMessageQueue(wg *sync.WaitGroup) {
 		logger.Log(0, "could not subscribe to message queue ...")
 		return
 	}
-	if token := client.Subscribe("ping/#", 0, mq.Ping); token.Wait() && token.Error() != nil {
+	if token := client.Subscribe("ping/#", 2, mq.Ping); token.Wait() && token.Error() != nil {
 		client.Disconnect(240)
 		logger.Log(0, "ping sub failed")
 	}
@@ -207,8 +224,10 @@ func runMessageQueue(wg *sync.WaitGroup) {
 		client.Disconnect(240)
 		logger.Log(0, "metrics sub failed")
 	}
-	for {
-	}
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
+	<-quit
+	logger.Log(0, "Message Queue shutting down")
 }
 
 func authServerUnaryInterceptor() grpc.ServerOption {

models/mqtt.go

@@ -1,10 +1,9 @@
 package models
 
-import "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
-
 type PeerUpdate struct {
-	Network string
-	Peers   []wgtypes.Peer
+	Network  string
+	Nodes    []Node
+	ExtPeers []ExtPeersResponse
 }
 
 type KeyUpdate struct {

mq/mq.go

@@ -1,19 +1,15 @@
 package mq
 
 import (
-	"encoding/json"
 	"errors"
-	"log"
-	"net"
+	"fmt"
 	"strings"
-	"time"
 
 	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"
-	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
 var DefaultHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
@@ -22,20 +18,11 @@ var DefaultHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Messa
 
 var Metrics mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 	logger.Log(0, "Metrics Handler")
+	//TODOD -- handle metrics data ---- store to database?
 }
 
 var Ping mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
-	logger.Log(0, "Ping Handler")
-	//test code --- create a node if it doesn't exit for testing only
-	createnode := models.Node{PublicKey: "DM5qhLAE20PG9BbfBCger+Ac9D2NDOwCtY1rbYDLf34=", Name: "testnode",
-		Endpoint: "10.0.0.1", MacAddress: "01:02:03:04:05:06", Password: "password", Network: "skynet"}
-	if _, err := logic.GetNode("01:02:03:04:05:06", "skynet"); err != nil {
-		err := logic.CreateNode(&createnode)
-		if err != nil {
-			log.Println(err)
-		}
-	}
-	//end of test code
+	logger.Log(0, "Ping Handler: "+msg.Topic())
 	go func() {
 		mac, net, err := GetMacNetwork(msg.Topic())
 		if err != nil {
@@ -56,6 +43,7 @@ var Ping mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 			return
 		}
 		node.SetLastCheckIn()
+		logger.Log(0, "ping processed")
 		// --TODO --set client version once feature is implemented.
 		//node.SetClientVersion(msg.Payload())
 	}()
@@ -76,7 +64,9 @@ var PublicKeyUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Mess
 		}
 		node.PublicKey = key
 		node.SetLastCheckIn()
-		UpdatePeers(&node, client)
+		if err := UpdatePeers(client, node); err != nil {
+			logger.Log(0, "error updating peers "+err.Error())
+		}
 	}()
 }
 
@@ -97,58 +87,63 @@ var IPUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
 		}
 		node.Endpoint = ip
 		node.SetLastCheckIn()
-		UpdatePeers(&node, client)
+		if err := UpdatePeers(client, node); err != nil {
+			logger.Log(0, "error updating peers "+err.Error())
+		}
 	}()
 }
 
-func UpdatePeers(node *models.Node, client mqtt.Client) {
-	peersToUpdate, err := logic.GetPeers(node)
+func UpdatePeers(client mqtt.Client, node models.Node) error {
+	var peerUpdate models.PeerUpdate
+	peerUpdate.Network = node.Network
+
+	nodes, err := logic.GetNetworkNodes(node.Network)
 	if err != nil {
-		logger.Log(0, "error retrieving peers")
-		return
+		return fmt.Errorf("unable to get network nodes %v: ", err)
 	}
-	for _, peerToUpdate := range peersToUpdate {
-		var peerUpdate models.PeerUpdate
-		peerUpdate.Network = node.Network
-
-		myPeers, err := logic.GetPeers(&peerToUpdate)
-		if err != nil {
-			logger.Log(0, "uable to get peers "+err.Error())
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	for _, peer := range nodes {
+		//don't need to update the initiatiing client
+		if peer.ID == node.ID {
 			continue
 		}
-		for i, myPeer := range myPeers {
-			var allowedIPs []net.IPNet
-			var allowedIP net.IPNet
-			endpoint, err := net.ResolveUDPAddr("udp", myPeer.Address+":"+string(myPeer.ListenPort))
-			if err != nil {
-				logger.Log(0, "error setting endpoint for peer "+err.Error())
-			}
-			for _, ipString := range myPeer.AllowedIPs {
-				_, ipNet, _ := net.ParseCIDR(ipString)
-				allowedIP = *ipNet
-				allowedIPs = append(allowedIPs, allowedIP)
-			}
-			key, err := wgtypes.ParseKey(myPeer.PublicKey)
-			if err != nil {
-				logger.Log(0, "err parsing publickey")
-				continue
-			}
-			peerUpdate.Peers[i].PublicKey = key
-			peerUpdate.Peers[i].Endpoint = endpoint
-			peerUpdate.Peers[i].PersistentKeepaliveInterval = time.Duration(myPeer.PersistentKeepalive)
-			peerUpdate.Peers[i].AllowedIPs = allowedIPs
-			peerUpdate.Peers[i].ProtocolVersion = 0
-		}
-		//PublishPeerUpdate(my)
-		data, err := json.Marshal(peerUpdate)
+		peerUpdate.Nodes = append(peerUpdate.Nodes, peer)
+		peerUpdate.ExtPeers, err = logic.GetExtPeersList(node.MacAddress, node.Network)
 		if err != nil {
-			logger.Log(0, "err marshalling data for peer update "+err.Error())
+			logger.Log(0)
+		}
+		if token := client.Publish("update/peers/"+peer.ID, 0, false, nodes); token.Wait() && token.Error() != nil {
+			logger.Log(0, "error publishing peer update "+peer.ID+" "+token.Error().Error())
+		}
+	}
+
+	return nil
+}
+
+func UpdateLocalPeers(client mqtt.Client, node models.Node) error {
+	nodes, err := logic.GetNetworkNodes(node.Network)
+	if err != nil {
+		return fmt.Errorf("unable to get network nodes %v: ", err)
+	}
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		return token.Error()
+	}
+	for _, peer := range nodes {
+		//don't need to update the initiatiing client
+		if peer.ID == node.ID {
+			continue
 		}
-		if token := client.Publish("update/peers/"+peerToUpdate.ID, 0, false, data); token.Wait() && token.Error() != nil {
-			logger.Log(0, "error publishing peer update "+token.Error().Error())
+		//if peer.Endpoint is on same lan as node.LocalAddress
+		//if TODO{
+		//continue
+		//}
+		if token := client.Publish("update/peers/"+peer.ID, 0, false, nodes); token.Wait() && token.Error() != nil {
+			logger.Log(0, "error publishing peer update "+peer.ID+" "+token.Error().Error())
 		}
-		client.Disconnect(250)
 	}
+	return nil
 }
 
 var LocalAddressUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
@@ -167,6 +162,9 @@ var LocalAddressUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.M
 		}
 		node.LocalAddress = string(msg.Payload())
 		node.SetLastCheckIn()
+		if err := UpdateLocalPeers(client, node); err != nil {
+			logger.Log(0, "error updating peers "+err.Error())
+		}
 	}()
 }
 
@@ -176,7 +174,7 @@ func GetMacNetwork(topic string) (string, string, error) {
 	if count == 1 {
 		return "", "", errors.New("invalid topic")
 	}
-	macnet := strings.Split(parts[count-1], "---")
+	macnet := strings.Split(parts[count-1], "-")
 	if len(macnet) != 2 {
 		return "", "", errors.New("topic id not in mac---network format")
 	}
@@ -189,7 +187,7 @@ func GetID(topic string) (string, error) {
 	if count == 1 {
 		return "", errors.New("invalid topic")
 	}
-	macnet := strings.Split(parts[count-1], "---")
+	macnet := strings.Split(parts[count-1], "-")
 	if len(macnet) != 2 {
 		return "", errors.New("topic id not in mac---network format")
 	}

netclient/functions/daemon.go

@@ -154,9 +154,19 @@ var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
 		var cfg config.ClientConfig
 		cfg.Network = peerUpdate.Network
 		cfg.ReadConfig()
-		err = wireguard.UpdateWgPeers(cfg.Node.Interface, peerUpdate.Peers)
+		peers, err := CalculatePeers(cfg.Node, peerUpdate.Nodes, cfg.Node.IsDualStack, cfg.Node.IsEgressGateway, cfg.Node.IsServer)
 		if err != nil {
-			ncutils.Log("error updating peers" + err.Error())
+			ncutils.Log("error calculating Peers " + err.Error())
+			return
+		}
+		extpeers, err := CalculateExtPeers(cfg.Node, peerUpdate.ExtPeers)
+		if err != nil {
+			ncutils.Log("error updated external wireguard peers " + err.Error())
+		}
+		peers = append(peers, extpeers...)
+		err = wireguard.UpdateWgPeers(cfg.Node.Interface, peers)
+		if err != nil {
+			ncutils.Log("error updating wireguard peers" + err.Error())
 			return
 		}
 		// path hardcoded for now... should be updated
@@ -187,7 +197,9 @@ func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) (*config.ClientCon
 		client.Disconnect(250)
 		return cfg, err
 	}
-	client.Disconnect(250)
+	if err := config.ModConfig(&cfg.Node); err != nil {
+		ncutils.Log("error updating local config " + err.Error())
+	}
 	return cfg, nil
 }
 
@@ -202,6 +214,11 @@ func Checkin(ctx context.Context, cfg config.ClientConfig, network string) {
 			//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
 		case <-time.After(time.Second * 10):
 			ncutils.Log("Checkin running")
+			//read latest config
+			cfg.ReadConfig()
+			//fix NodeID to remove ### so NodeID can be used as message topic
+			//remove with GRA-73
+			cfg.Node.ID = strings.Replace(cfg.Node.ID, "###", "-", 1)
 			if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
 				extIP, err := ncutils.GetPublicIP()
 				if err != nil {
@@ -242,6 +259,10 @@ func UpdateEndpoint(cfg config.ClientConfig, network, ip string) {
 	if token := client.Publish("update/ip/"+cfg.Node.ID, 0, false, ip); token.Wait() && token.Error() != nil {
 		ncutils.Log("error publishing endpoint update " + token.Error().Error())
 	}
+	cfg.Node.Endpoint = ip
+	if err := config.Write(&cfg, cfg.Network); err != nil {
+		ncutils.Log("error updating local config " + err.Error())
+	}
 	client.Disconnect(250)
 }
 
@@ -252,13 +273,19 @@ func UpdateLocalAddress(cfg config.ClientConfig, network, ip string) {
 	if token := client.Publish("update/localaddress/"+cfg.Node.ID, 0, false, ip); token.Wait() && token.Error() != nil {
 		ncutils.Log("error publishing local address update " + token.Error().Error())
 	}
+	cfg.Node.LocalAddress = ip
+	ncutils.Log("updating local address in local config to: " + cfg.Node.LocalAddress)
+	if err := config.Write(&cfg, cfg.Network); err != nil {
+		ncutils.Log("error updating local config " + err.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, 0, false, "hello world!"); token.Wait() && token.Error() != nil {
+	ncutils.Log("sending ping " + cfg.Node.ID)
+	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/functions/peers.go

@@ -0,0 +1,167 @@
+package functions
+
+import (
+	"log"
+	"net"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/gravitl/netmaker/models"
+	"github.com/gravitl/netmaker/netclient/ncutils"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+func CalculatePeers(thisNode models.Node, peernodes []models.Node, dualstack, egressgateway, server string) ([]wgtypes.Peer, error) {
+	//hasGateway := false
+	var gateways []string
+	var peers []wgtypes.Peer
+
+	keepalive := thisNode.PersistentKeepalive
+	keepalivedur, _ := time.ParseDuration(strconv.FormatInt(int64(keepalive), 10) + "s")
+	keepaliveserver, err := time.ParseDuration(strconv.FormatInt(int64(5), 10) + "s")
+	if err != nil {
+		log.Fatalf("Issue with format of keepalive value. Please update netconfig: %v", err)
+	}
+	for _, node := range peernodes {
+		pubkey, err := wgtypes.ParseKey(node.PublicKey)
+		if err != nil {
+			log.Println("error parsing key")
+			//return peers, hasGateway, gateways, err
+		}
+
+		if thisNode.PublicKey == node.PublicKey {
+			continue
+		}
+		if thisNode.Endpoint == node.Endpoint {
+			if thisNode.LocalAddress != node.LocalAddress && node.LocalAddress != "" {
+				node.Endpoint = node.LocalAddress
+			} else {
+				continue
+			}
+		}
+
+		var peer wgtypes.Peer
+		var peeraddr = net.IPNet{
+			IP:   net.ParseIP(node.Address),
+			Mask: net.CIDRMask(32, 32),
+		}
+		var allowedips []net.IPNet
+		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(thisNode.LocalAddress)) { // ensuring egress gateway range does not contain public ip of node
+					ncutils.PrintLog("egress IP range of "+iprange+" overlaps with "+thisNode.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 == "yes" {
+			var addr6 = net.IPNet{
+				IP:   net.ParseIP(node.Address6),
+				Mask: net.CIDRMask(128, 128),
+			}
+			allowedips = append(allowedips, addr6)
+		}
+		if thisNode.IsServer == "yes" && !(node.IsServer == "yes") {
+			peer = wgtypes.Peer{
+				PublicKey:                   pubkey,
+				PersistentKeepaliveInterval: keepaliveserver,
+				AllowedIPs:                  allowedips,
+			}
+		} else if keepalive != 0 {
+			peer = wgtypes.Peer{
+				PublicKey:                   pubkey,
+				PersistentKeepaliveInterval: keepalivedur,
+				Endpoint: &net.UDPAddr{
+					IP:   net.ParseIP(node.Endpoint),
+					Port: int(node.ListenPort),
+				},
+				AllowedIPs: allowedips,
+			}
+		} else {
+			peer = wgtypes.Peer{
+				PublicKey: pubkey,
+				Endpoint: &net.UDPAddr{
+					IP:   net.ParseIP(node.Endpoint),
+					Port: int(node.ListenPort),
+				},
+				AllowedIPs: allowedips,
+			}
+		}
+		peers = append(peers, peer)
+	}
+	return peers, nil
+}
+
+func CalculateExtPeers(thisNode models.Node, extPeers []models.ExtPeersResponse) ([]wgtypes.Peer, error) {
+	var peers []wgtypes.Peer
+	var err error
+	for _, extPeer := range extPeers {
+		pubkey, err := wgtypes.ParseKey(extPeer.PublicKey)
+		if err != nil {
+			log.Println("error parsing key")
+			return peers, err
+		}
+
+		if thisNode.PublicKey == extPeer.PublicKey {
+			continue
+		}
+
+		var peer wgtypes.Peer
+		var peeraddr = net.IPNet{
+			IP:   net.ParseIP(extPeer.Address),
+			Mask: net.CIDRMask(32, 32),
+		}
+		var allowedips []net.IPNet
+		allowedips = append(allowedips, peeraddr)
+
+		if extPeer.Address6 != "" && thisNode.IsDualStack == "yes" {
+			var addr6 = net.IPNet{
+				IP:   net.ParseIP(extPeer.Address6),
+				Mask: net.CIDRMask(128, 128),
+			}
+			allowedips = append(allowedips, addr6)
+		}
+		peer = wgtypes.Peer{
+			PublicKey:  pubkey,
+			AllowedIPs: allowedips,
+		}
+		peers = append(peers, peer)
+	}
+	return peers, err
+}