mirror of
https://github.com/gravitl/netmaker.git
synced 2024-09-22 00:06:04 +08:00
369 lines
12 KiB
Go
369 lines
12 KiB
Go
package functions
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"runtime"
|
|
"sync"
|
|
"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"
|
|
)
|
|
|
|
var messageCache = make(map[string]string, 20)
|
|
|
|
const lastNodeUpdate = "lnu"
|
|
const lastPeerUpdate = "lpu"
|
|
|
|
func insert(network, which, cache string) {
|
|
var mu sync.Mutex
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
messageCache[fmt.Sprintf("%s%s", network, which)] = cache
|
|
}
|
|
|
|
func read(network, which string) string {
|
|
return messageCache[fmt.Sprintf("%s%s", network, which)]
|
|
}
|
|
|
|
// 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()
|
|
for _, server := range cfg.Node.NetworkSettings.DefaultServerAddrs {
|
|
if server.Address != "" && server.IsLeader {
|
|
ncutils.Log(fmt.Sprintf("adding server (%s) to listen on network %s \n", server.Address, cfg.Node.Network))
|
|
opts.AddBroker(server.Address + ":1883")
|
|
break
|
|
}
|
|
}
|
|
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, mqtt.MessageHandler(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, mqtt.MessageHandler(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
|
|
func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
|
|
//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
|
|
}
|
|
ncutils.Log("received message to update node " + newNode.Name)
|
|
// see if cache hit, if so skip
|
|
var currentMessage = read(newNode.Network, lastNodeUpdate)
|
|
if currentMessage == string(msg.Payload()) {
|
|
return
|
|
}
|
|
insert(newNode.Network, lastNodeUpdate, string(msg.Payload()))
|
|
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
|
|
func UpdatePeers(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
|
|
}
|
|
// see if cache hit, if so skip
|
|
var currentMessage = read(peerUpdate.Network, lastPeerUpdate)
|
|
if currentMessage == string(msg.Payload()) {
|
|
return
|
|
}
|
|
insert(peerUpdate.Network, lastPeerUpdate, string(msg.Payload()))
|
|
ncutils.Log("update peer handler")
|
|
var cfg config.ClientConfig
|
|
cfg.Network = peerUpdate.Network
|
|
cfg.ReadConfig()
|
|
var shouldReSub = shouldResub(cfg.Node.NetworkSettings.DefaultServerAddrs, peerUpdate.ServerAddrs)
|
|
if shouldReSub {
|
|
Resubscribe(client, &cfg)
|
|
cfg.Node.NetworkSettings.DefaultServerAddrs = peerUpdate.ServerAddrs
|
|
}
|
|
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
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Resubscribe --- handles resubscribing if needed
|
|
func Resubscribe(client mqtt.Client, cfg *config.ClientConfig) error {
|
|
if err := config.ModConfig(&cfg.Node); err == nil {
|
|
ncutils.Log("resubbing on network " + cfg.Node.Network)
|
|
client.Disconnect(250)
|
|
client = SetupMQTT(cfg)
|
|
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())
|
|
}
|
|
ncutils.Log("finished re subbing")
|
|
return nil
|
|
} else {
|
|
ncutils.Log("could not mod config when re-subbing")
|
|
return err
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
func shouldResub(currentServers, newServers []models.ServerAddr) bool {
|
|
if len(currentServers) != len(newServers) {
|
|
return true
|
|
}
|
|
for _, srv := range currentServers {
|
|
if !ncutils.ServerAddrSliceContains(newServers, srv) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|