diff --git a/main.go b/main.go index 52243051..3d527a0b 100644 --- a/main.go +++ b/main.go @@ -9,7 +9,7 @@ import ( "runtime/debug" "strconv" "sync" - "time" + "syscall" mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/gravitl/netmaker/auth" @@ -118,16 +118,16 @@ 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) - } - } + //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() } @@ -204,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") } @@ -224,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 { diff --git a/models/mqtt.go b/models/mqtt.go index da7265f3..07bb2157 100644 --- a/models/mqtt.go +++ b/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 { diff --git a/mq/mq.go b/mq/mq.go index 3620b215..fd9657a7 100644 --- a/mq/mq.go +++ b/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) - if err != nil { - logger.Log(0, "error retrieving peers") - return - } - for _, peerToUpdate := range peersToUpdate { - var peerUpdate models.PeerUpdate - peerUpdate.Network = node.Network +func UpdatePeers(client mqtt.Client, node models.Node) error { + 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()) + 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 } - 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/"+peerToUpdate.ID, 0, false, data); token.Wait() && token.Error() != nil { - logger.Log(0, "error publishing peer update "+token.Error().Error()) + 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 +} + +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 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()) + } + } + 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") } diff --git a/netclient/functions/daemon.go b/netclient/functions/daemon.go index a5360319..6e212c77 100644 --- a/netclient/functions/daemon.go +++ b/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) diff --git a/netclient/functions/peers.go b/netclient/functions/peers.go new file mode 100644 index 00000000..833ebc90 --- /dev/null +++ b/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 +}