mirror of
https://github.com/gravitl/netmaker.git
synced 2025-10-05 08:47:35 +08:00
subscribed message handlers implemented
This commit is contained in:
1
go.mod
1
go.mod
@@ -49,4 +49,5 @@ require (
|
|||||||
github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
|
github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
|
||||||
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
|
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
|
||||||
google.golang.org/appengine v1.4.0 // indirect
|
google.golang.org/appengine v1.4.0 // indirect
|
||||||
|
gopkg.in/ini.v1 v1.66.2 // indirect
|
||||||
)
|
)
|
||||||
|
2
go.sum
2
go.sum
@@ -292,6 +292,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 h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
|
||||||
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
|
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/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.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||||
gopkg.in/yaml.v2 v2.2.3/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=
|
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
|
@@ -6,7 +6,6 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"runtime"
|
|
||||||
"strings"
|
"strings"
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
@@ -68,8 +67,7 @@ func Netclient(ctx context.Context, network string) {
|
|||||||
}
|
}
|
||||||
client.AddRoute("update/"+cfg.Node.ID, NodeUpdate)
|
client.AddRoute("update/"+cfg.Node.ID, NodeUpdate)
|
||||||
client.AddRoute("update/peers/"+cfg.Node.ID, UpdatePeers)
|
client.AddRoute("update/peers/"+cfg.Node.ID, UpdatePeers)
|
||||||
//handle key updates in node update
|
client.AddRoute("update/keys/"+cfg.Node.ID, UpdateKeys)
|
||||||
//client.AddRoute("update/keys/"+cfg.Node.ID, UpdateKeys)
|
|
||||||
defer client.Disconnect(250)
|
defer client.Disconnect(250)
|
||||||
go Checkin(ctx, cfg, network)
|
go Checkin(ctx, cfg, network)
|
||||||
go Metrics(ctx, cfg, network)
|
go Metrics(ctx, cfg, network)
|
||||||
@@ -88,85 +86,19 @@ var NodeUpdate mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
|
|||||||
ncutils.Log("received message to update node " + string(msg.Payload()))
|
ncutils.Log("received message to update node " + string(msg.Payload()))
|
||||||
//potentiall blocking i/o so do this in a go routine
|
//potentiall blocking i/o so do this in a go routine
|
||||||
go func() {
|
go func() {
|
||||||
var newNode models.Node
|
var data models.Node
|
||||||
var cfg config.ClientConfig
|
err := json.Unmarshal(msg.Payload(), &data)
|
||||||
cfg.Network = newNode.Network
|
|
||||||
cfg.ReadConfig()
|
|
||||||
err := json.Unmarshal(msg.Payload(), &newNode)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
ncutils.Log("error unmarshalling node update data" + err.Error())
|
ncutils.Log("error unmarshalling node update data" + err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
//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
|
|
||||||
}
|
|
||||||
case models.NODE_UPDATE_KEY:
|
|
||||||
UpdateKeys(&cfg, client)
|
|
||||||
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
|
|
||||||
}
|
|
||||||
if err := wireguard.UpdateWgInterface(cfg.Node.Interface, privateKey, nameserver, newNode); err != nil {
|
|
||||||
ncutils.Log("error updating wireguard config " + err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// path hardcoded for now... should be updated
|
|
||||||
err = wireguard.ApplyWGQuickConf("/etc/netclient/config/" + cfg.Node.Interface + ".conf")
|
|
||||||
if err != nil {
|
|
||||||
ncutils.Log("error restarting wg after node update " + err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// UpdatePeers -- mqtt message handler for /update/peers/<NodeID> topic
|
|
||||||
var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
|
||||||
ncutils.Log("received message to update peers " + string(msg.Payload()))
|
|
||||||
go func() {
|
|
||||||
var peerUpdate models.PeerUpdate
|
|
||||||
err := json.Unmarshal(msg.Payload(), &peerUpdate)
|
|
||||||
if err != nil {
|
|
||||||
ncutils.Log("error unmarshalling peer data")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
var cfg config.ClientConfig
|
var cfg config.ClientConfig
|
||||||
cfg.Network = peerUpdate.Network
|
cfg.Network = data.Network
|
||||||
cfg.ReadConfig()
|
cfg.ReadConfig()
|
||||||
peers, err := CalculatePeers(cfg.Node, peerUpdate.Nodes, cfg.Node.IsDualStack, cfg.Node.IsEgressGateway, cfg.Node.IsServer)
|
nameserver := cfg.Server.CoreDNSAddr
|
||||||
if err != nil {
|
privateKey, err := wireguard.RetrievePrivKey(data.Network)
|
||||||
ncutils.Log("error calculating Peers " + err.Error())
|
if err := wireguard.UpdateWgInterface(cfg.Node.Interface, privateKey, nameserver, data); err != nil {
|
||||||
return
|
ncutils.Log("error updating wireguard config " + err.Error())
|
||||||
}
|
|
||||||
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
|
return
|
||||||
}
|
}
|
||||||
// path hardcoded for now... should be updated
|
// path hardcoded for now... should be updated
|
||||||
@@ -178,29 +110,62 @@ var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message)
|
|||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpdateKeys -- updates private key and returns new publickey
|
// UpdatePeers -- mqtt message handler for /update/peers/<NodeID> topic
|
||||||
func UpdateKeys(cfg *config.ClientConfig, client mqtt.Client) (*config.ClientConfig, error) {
|
var UpdatePeers mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
||||||
ncutils.Log("received message to update keys")
|
ncutils.Log("received message to update peers " + string(msg.Payload()))
|
||||||
//potentiall blocking i/o so do this in a go routine
|
//potentiall blocking i/o so do this in a go routine
|
||||||
key, err := wgtypes.GeneratePrivateKey()
|
go func() {
|
||||||
if err != nil {
|
var peerUpdate models.PeerUpdate
|
||||||
ncutils.Log("error generating privatekey " + err.Error())
|
err := json.Unmarshal(msg.Payload(), &peerUpdate)
|
||||||
return cfg, err
|
if err != nil {
|
||||||
}
|
ncutils.Log("error unmarshalling peer data")
|
||||||
if err := wireguard.UpdatePrivateKey(cfg.Node.Interface, key.String()); err != nil {
|
return
|
||||||
ncutils.Log("error updating wireguard key " + err.Error())
|
}
|
||||||
return cfg, err
|
var cfg config.ClientConfig
|
||||||
}
|
cfg.Network = peerUpdate.Network
|
||||||
publicKey := key.PublicKey()
|
cfg.ReadConfig()
|
||||||
if token := client.Publish("update/publickey/"+cfg.Node.ID, 0, false, publicKey.String()); token.Wait() && token.Error() != nil {
|
err = wireguard.UpdateWgPeers(cfg.Node.Interface, peerUpdate.Peers)
|
||||||
ncutils.Log("error publishing publickey update " + token.Error().Error())
|
if err != nil {
|
||||||
|
ncutils.Log("error updating peers" + err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// path hardcoded for now... should be updated
|
||||||
|
err = wireguard.ApplyWGQuickConf("/etc/netclient/config/" + cfg.Node.Interface + ".conf")
|
||||||
|
if err != nil {
|
||||||
|
ncutils.Log("error restarting wg after peer update " + err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
// UpdateKeys -- mqtt message handler for /update/keys/<NodeID> topic
|
||||||
|
var UpdateKeys mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
|
||||||
|
ncutils.Log("received message to update keys " + string(msg.Payload()))
|
||||||
|
//potentiall blocking i/o so do this in a go routine
|
||||||
|
go func() {
|
||||||
|
var data models.KeyUpdate
|
||||||
|
if err := json.Unmarshal(msg.Payload(), &data); err != nil {
|
||||||
|
ncutils.Log("error unmarshalling key update data" + err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var cfg config.ClientConfig
|
||||||
|
cfg.Network = data.Network
|
||||||
|
cfg.ReadConfig()
|
||||||
|
key, err := wgtypes.GeneratePrivateKey()
|
||||||
|
if err != nil {
|
||||||
|
ncutils.Log("error generating privatekey " + err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := wireguard.UpdatePrivateKey(data.Interface, key.String()); err != nil {
|
||||||
|
ncutils.Log("error updating wireguard key " + err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
|
publicKey := key.PublicKey()
|
||||||
|
if token := client.Publish("update/publickey/"+cfg.Node.ID, 0, false, publicKey.String()); token.Wait() && token.Error() != nil {
|
||||||
|
ncutils.Log("error publishing publickey update " + token.Error().Error())
|
||||||
|
}
|
||||||
client.Disconnect(250)
|
client.Disconnect(250)
|
||||||
return cfg, err
|
}()
|
||||||
}
|
|
||||||
if err := config.ModConfig(&cfg.Node); err != nil {
|
|
||||||
ncutils.Log("error updating local config " + err.Error())
|
|
||||||
}
|
|
||||||
return cfg, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Checkin -- go routine that checks for public or local ip changes, publishes changes
|
// Checkin -- go routine that checks for public or local ip changes, publishes changes
|
||||||
@@ -214,11 +179,6 @@ func Checkin(ctx context.Context, cfg config.ClientConfig, network string) {
|
|||||||
//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
|
//delay should be configuraable -> use cfg.Node.NetworkSettings.DefaultCheckInInterval ??
|
||||||
case <-time.After(time.Second * 10):
|
case <-time.After(time.Second * 10):
|
||||||
ncutils.Log("Checkin running")
|
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" {
|
if cfg.Node.Roaming == "yes" && cfg.Node.IsStatic != "yes" {
|
||||||
extIP, err := ncutils.GetPublicIP()
|
extIP, err := ncutils.GetPublicIP()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -259,10 +219,6 @@ 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 {
|
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())
|
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)
|
client.Disconnect(250)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -273,19 +229,13 @@ 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 {
|
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())
|
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)
|
client.Disconnect(250)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Hello -- ping the broker to let server know node is alive and doing fine
|
// Hello -- ping the broker to let server know node is alive and doing fine
|
||||||
func Hello(cfg config.ClientConfig, network string) {
|
func Hello(cfg config.ClientConfig, network string) {
|
||||||
client := SetupMQTT(cfg)
|
client := SetupMQTT(cfg)
|
||||||
ncutils.Log("sending ping " + cfg.Node.ID)
|
if token := client.Publish("ping/"+cfg.Node.ID, 0, false, "hello world!"); token.Wait() && token.Error() != nil {
|
||||||
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())
|
ncutils.Log("error publishing ping " + token.Error().Error())
|
||||||
}
|
}
|
||||||
client.Disconnect(250)
|
client.Disconnect(250)
|
||||||
|
Reference in New Issue
Block a user