mirror of
https://github.com/gravitl/netmaker.git
synced 2024-09-23 00:36:40 +08:00
440 lines
14 KiB
Go
440 lines
14 KiB
Go
package functions
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
"github.com/gravitl/netmaker/logger"
|
|
"github.com/gravitl/netmaker/models"
|
|
"github.com/gravitl/netmaker/mq"
|
|
"github.com/gravitl/netmaker/netclient/auth"
|
|
"github.com/gravitl/netmaker/netclient/config"
|
|
"github.com/gravitl/netmaker/netclient/global_settings"
|
|
"github.com/gravitl/netmaker/netclient/local"
|
|
"github.com/gravitl/netmaker/netclient/ncutils"
|
|
"github.com/gravitl/netmaker/netclient/wireguard"
|
|
|
|
nmproxy "github.com/gravitl/netmaker/nm-proxy"
|
|
"github.com/gravitl/netmaker/nm-proxy/manager"
|
|
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
|
|
)
|
|
|
|
var ProxyMgmChan = make(chan *manager.ManagerAction, 100)
|
|
var messageCache = new(sync.Map)
|
|
var ProxyStatus = "OFF"
|
|
var serverSet map[string]bool
|
|
|
|
var mqclient mqtt.Client
|
|
|
|
const lastNodeUpdate = "lnu"
|
|
const lastPeerUpdate = "lpu"
|
|
|
|
type cachedMessage struct {
|
|
Message string
|
|
LastSeen time.Time
|
|
}
|
|
|
|
// Daemon runs netclient daemon from command line
|
|
func Daemon() error {
|
|
logger.Log(0, "netclient daemon started -- version:", ncutils.Version)
|
|
UpdateClientConfig()
|
|
if err := ncutils.SavePID(); err != nil {
|
|
return err
|
|
}
|
|
// reference required to eliminate unused statticcheck
|
|
serverSet = make(map[string]bool)
|
|
serverSet["dummy"] = false
|
|
// set ipforwarding on startup
|
|
err := local.SetIPForwarding()
|
|
if err != nil {
|
|
logger.Log(0, err.Error())
|
|
}
|
|
|
|
// == add waitgroup and cancel for checkin routine ==
|
|
wg := sync.WaitGroup{}
|
|
quit := make(chan os.Signal, 1)
|
|
reset := make(chan os.Signal, 1)
|
|
signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
|
|
signal.Notify(reset, syscall.SIGHUP)
|
|
cancel := startGoRoutines(&wg)
|
|
for {
|
|
select {
|
|
case <-quit:
|
|
cancel()
|
|
logger.Log(0, "shutting down netclient daemon")
|
|
wg.Wait()
|
|
if mqclient != nil {
|
|
mqclient.Disconnect(250)
|
|
}
|
|
logger.Log(0, "shutdown complete")
|
|
return nil
|
|
case <-reset:
|
|
logger.Log(0, "received reset")
|
|
cancel()
|
|
wg.Wait()
|
|
if mqclient != nil {
|
|
mqclient.Disconnect(250)
|
|
}
|
|
logger.Log(0, "restarting daemon")
|
|
cancel = startGoRoutines(&wg)
|
|
}
|
|
}
|
|
}
|
|
|
|
func startGoRoutines(wg *sync.WaitGroup) context.CancelFunc {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
serverSet := make(map[string]bool)
|
|
networks, _ := ncutils.GetSystemNetworks()
|
|
for _, network := range networks {
|
|
logger.Log(3, "initializing network", network)
|
|
cfg := config.ClientConfig{}
|
|
cfg.Network = network
|
|
cfg.ReadConfig()
|
|
if cfg.Node.Connected == "yes" {
|
|
if err := wireguard.ApplyConf(&cfg.Node, cfg.Node.Interface, ncutils.GetNetclientPathSpecific()+cfg.Node.Interface+".conf"); err != nil {
|
|
logger.Log(0, "failed to start ", cfg.Node.Interface, "wg interface", err.Error())
|
|
}
|
|
if cfg.PublicIPService != "" {
|
|
global_settings.PublicIPServices[network] = cfg.PublicIPService
|
|
}
|
|
}
|
|
|
|
server := cfg.Server.Server
|
|
if !serverSet[server] {
|
|
// == subscribe to all nodes for each on machine ==
|
|
serverSet[server] = true
|
|
logger.Log(1, "started daemon for server ", server)
|
|
local.SetNetmakerDomainRoute(cfg.Server.API)
|
|
wg.Add(1)
|
|
go messageQueue(ctx, wg, &cfg)
|
|
}
|
|
}
|
|
wg.Add(1)
|
|
go Checkin(ctx, wg)
|
|
|
|
if len(networks) != 0 {
|
|
go func() {
|
|
cfg := config.ClientConfig{}
|
|
cfg.Network = networks[0]
|
|
cfg.ReadConfig()
|
|
apiHost, _, err := net.SplitHostPort(cfg.Server.API)
|
|
if err == nil {
|
|
if ProxyStatus != "ON" {
|
|
ProxyStatus = "ON"
|
|
pCtx, pCancel := context.WithCancel(context.Background())
|
|
go nmproxy.Start(pCtx, ProxyMgmChan, apiHost)
|
|
quit := make(chan os.Signal, 1)
|
|
signal.Notify(quit, syscall.SIGTERM, os.Interrupt)
|
|
<-quit
|
|
pCancel()
|
|
logger.Log(0, "Proxy Shutting down....")
|
|
}
|
|
|
|
}
|
|
}()
|
|
|
|
}
|
|
|
|
go func(networks []string) {
|
|
|
|
for _, network := range networks {
|
|
logger.Log(0, "Collecting interface and peers info to configure proxy...")
|
|
cfg := config.ClientConfig{}
|
|
cfg.Network = network
|
|
cfg.ReadConfig()
|
|
node, err := GetNodeInfo(&cfg)
|
|
if err != nil {
|
|
log.Println("Failed to get node info: ", err)
|
|
continue
|
|
}
|
|
ProxyMgmChan <- &manager.ManagerAction{
|
|
Action: manager.AddInterface,
|
|
Payload: manager.ManagerPayload{
|
|
InterfaceName: node.Node.Interface,
|
|
Peers: node.Peers,
|
|
},
|
|
}
|
|
|
|
}
|
|
|
|
}(networks)
|
|
return cancel
|
|
}
|
|
func GetNodeInfo(cfg *config.ClientConfig) (models.NodeGet, error) {
|
|
var nodeGET models.NodeGet
|
|
token, err := Authenticate(cfg)
|
|
if err != nil {
|
|
return nodeGET, err
|
|
}
|
|
url := fmt.Sprintf("https://%s/api/nodes/%s/%s", cfg.Server.API, cfg.Network, cfg.Node.ID)
|
|
response, err := API("", http.MethodGet, url, token)
|
|
if err != nil {
|
|
return nodeGET, err
|
|
}
|
|
if response.StatusCode != http.StatusOK {
|
|
bytes, err := io.ReadAll(response.Body)
|
|
if err != nil {
|
|
fmt.Println(err)
|
|
}
|
|
return nodeGET, (fmt.Errorf("%s %w", string(bytes), err))
|
|
}
|
|
defer response.Body.Close()
|
|
if err := json.NewDecoder(response.Body).Decode(&nodeGET); err != nil {
|
|
return nodeGET, fmt.Errorf("error decoding node %w", err)
|
|
}
|
|
return nodeGET, nil
|
|
}
|
|
|
|
// UpdateKeys -- updates private key and returns new publickey
|
|
func UpdateKeys(nodeCfg *config.ClientConfig, client mqtt.Client) error {
|
|
logger.Log(0, "interface:", nodeCfg.Node.Interface, "received message to update wireguard keys for network ", nodeCfg.Network)
|
|
key, err := wgtypes.GeneratePrivateKey()
|
|
if err != nil {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, "error generating privatekey ", err.Error())
|
|
return err
|
|
}
|
|
file := ncutils.GetNetclientPathSpecific() + nodeCfg.Node.Interface + ".conf"
|
|
if err := wireguard.UpdatePrivateKey(file, key.String()); err != nil {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, "error updating wireguard key ", err.Error())
|
|
return err
|
|
}
|
|
if storeErr := wireguard.StorePrivKey(key.String(), nodeCfg.Network); storeErr != nil {
|
|
logger.Log(0, "network:", nodeCfg.Network, "failed to save private key", storeErr.Error())
|
|
return storeErr
|
|
}
|
|
|
|
nodeCfg.Node.PublicKey = key.PublicKey().String()
|
|
PublishNodeUpdate(nodeCfg)
|
|
return nil
|
|
}
|
|
|
|
// == Private ==
|
|
|
|
// sets MQ client subscriptions for a specific node config
|
|
// should be called for each node belonging to a given server
|
|
func setSubscriptions(client mqtt.Client, nodeCfg *config.ClientConfig) {
|
|
if token := client.Subscribe(fmt.Sprintf("update/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(NodeUpdate)); token.WaitTimeout(mq.MQ_TIMEOUT*time.Second) && token.Error() != nil {
|
|
if token.Error() == nil {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, "connection timeout")
|
|
} else {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, token.Error().Error())
|
|
}
|
|
return
|
|
}
|
|
if token := client.Subscribe(fmt.Sprintf("proxy/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(ProxyUpdate)); token.WaitTimeout(mq.MQ_TIMEOUT*time.Second) && token.Error() != nil {
|
|
if token.Error() == nil {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, "connection timeout")
|
|
} else {
|
|
logger.Log(0, "network:", nodeCfg.Node.Network, token.Error().Error())
|
|
}
|
|
return
|
|
}
|
|
logger.Log(3, fmt.Sprintf("subscribed to node updates for node %s update/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
|
|
if token := client.Subscribe(fmt.Sprintf("peers/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(UpdatePeers)); token.Wait() && token.Error() != nil {
|
|
logger.Log(0, "network", nodeCfg.Node.Network, token.Error().Error())
|
|
return
|
|
}
|
|
logger.Log(3, fmt.Sprintf("subscribed to peer updates for node %s peers/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
|
|
}
|
|
|
|
// on a delete usually, pass in the nodecfg to unsubscribe client broker communications
|
|
// for the node in nodeCfg
|
|
func unsubscribeNode(client mqtt.Client, nodeCfg *config.ClientConfig) {
|
|
client.Unsubscribe(fmt.Sprintf("update/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID))
|
|
var ok = true
|
|
if token := client.Unsubscribe(fmt.Sprintf("update/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID)); token.WaitTimeout(mq.MQ_TIMEOUT*time.Second) && token.Error() != nil {
|
|
if token.Error() == nil {
|
|
logger.Log(1, "network:", nodeCfg.Node.Network, "unable to unsubscribe from updates for node ", nodeCfg.Node.Name, "\n", "connection timeout")
|
|
} else {
|
|
logger.Log(1, "network:", nodeCfg.Node.Network, "unable to unsubscribe from updates for node ", nodeCfg.Node.Name, "\n", token.Error().Error())
|
|
}
|
|
ok = false
|
|
}
|
|
if token := client.Unsubscribe(fmt.Sprintf("peers/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID)); token.WaitTimeout(mq.MQ_TIMEOUT*time.Second) && token.Error() != nil {
|
|
if token.Error() == nil {
|
|
logger.Log(1, "network:", nodeCfg.Node.Network, "unable to unsubscribe from peer updates for node", nodeCfg.Node.Name, "\n", "connection timeout")
|
|
} else {
|
|
logger.Log(1, "network:", nodeCfg.Node.Network, "unable to unsubscribe from peer updates for node", nodeCfg.Node.Name, "\n", token.Error().Error())
|
|
}
|
|
ok = false
|
|
}
|
|
if ok {
|
|
logger.Log(1, "network:", nodeCfg.Node.Network, "successfully unsubscribed node ", nodeCfg.Node.ID, " : ", nodeCfg.Node.Name)
|
|
}
|
|
}
|
|
|
|
// sets up Message Queue and subsribes/publishes updates to/from server
|
|
// the client should subscribe to ALL nodes that exist on server locally
|
|
func messageQueue(ctx context.Context, wg *sync.WaitGroup, cfg *config.ClientConfig) {
|
|
defer wg.Done()
|
|
logger.Log(0, "network:", cfg.Node.Network, "netclient message queue started for server:", cfg.Server.Server)
|
|
err := setupMQTT(cfg)
|
|
if err != nil {
|
|
logger.Log(0, "unable to connect to broker", cfg.Server.Server, err.Error())
|
|
return
|
|
}
|
|
//defer mqclient.Disconnect(250)
|
|
<-ctx.Done()
|
|
logger.Log(0, "shutting down message queue for server", cfg.Server.Server)
|
|
}
|
|
|
|
// func setMQTTSingenton creates a connection to broker for single use (ie to publish a message)
|
|
// only to be called from cli (eg. connect/disconnect, join, leave) and not from daemon ---
|
|
func setupMQTTSingleton(cfg *config.ClientConfig) error {
|
|
opts := mqtt.NewClientOptions()
|
|
server := cfg.Server.Server
|
|
port := cfg.Server.MQPort
|
|
pass, err := os.ReadFile(ncutils.GetNetclientPathSpecific() + "secret-" + cfg.Network)
|
|
if err != nil {
|
|
return fmt.Errorf("could not read secrets file %w", err)
|
|
}
|
|
opts.AddBroker("mqtts://" + server + ":" + port)
|
|
opts.SetUsername(cfg.Node.ID)
|
|
opts.SetPassword(string(pass))
|
|
mqclient = mqtt.NewClient(opts)
|
|
var connecterr error
|
|
opts.SetClientID(ncutils.MakeRandomString(23))
|
|
if token := mqclient.Connect(); !token.WaitTimeout(30*time.Second) || token.Error() != nil {
|
|
logger.Log(0, "unable to connect to broker, retrying ...")
|
|
if token.Error() == nil {
|
|
connecterr = errors.New("connect timeout")
|
|
} else {
|
|
connecterr = token.Error()
|
|
}
|
|
}
|
|
return connecterr
|
|
}
|
|
|
|
// setupMQTT creates a connection to broker and returns client
|
|
// this function is primarily used to create a connection to publish to the broker
|
|
func setupMQTT(cfg *config.ClientConfig) error {
|
|
opts := mqtt.NewClientOptions()
|
|
server := cfg.Server.Server
|
|
port := cfg.Server.MQPort
|
|
pass, err := os.ReadFile(ncutils.GetNetclientPathSpecific() + "secret-" + cfg.Network)
|
|
if err != nil {
|
|
return fmt.Errorf("could not read secrets file %w", err)
|
|
}
|
|
opts.AddBroker(fmt.Sprintf("mqtts://%s:%s", server, port))
|
|
opts.SetUsername(cfg.Node.ID)
|
|
opts.SetPassword(string(pass))
|
|
opts.SetClientID(ncutils.MakeRandomString(23))
|
|
opts.SetDefaultPublishHandler(All)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetConnectRetry(true)
|
|
opts.SetConnectRetryInterval(time.Second << 2)
|
|
opts.SetKeepAlive(time.Minute >> 1)
|
|
opts.SetWriteTimeout(time.Minute)
|
|
|
|
opts.SetOnConnectHandler(func(client mqtt.Client) {
|
|
networks, err := ncutils.GetSystemNetworks()
|
|
if err != nil {
|
|
logger.Log(0, "error retriving networks", err.Error())
|
|
}
|
|
for _, network := range networks {
|
|
var currNodeCfg config.ClientConfig
|
|
currNodeCfg.Network = network
|
|
currNodeCfg.ReadConfig()
|
|
setSubscriptions(client, &currNodeCfg)
|
|
}
|
|
})
|
|
opts.SetOrderMatters(true)
|
|
opts.SetResumeSubs(true)
|
|
opts.SetConnectionLostHandler(func(c mqtt.Client, e error) {
|
|
logger.Log(0, "network:", cfg.Node.Network, "detected broker connection lost for", cfg.Server.Server)
|
|
})
|
|
mqclient = mqtt.NewClient(opts)
|
|
var connecterr error
|
|
for count := 0; count < 3; count++ {
|
|
connecterr = nil
|
|
if token := mqclient.Connect(); !token.WaitTimeout(30*time.Second) || token.Error() != nil {
|
|
logger.Log(0, "unable to connect to broker, retrying ...")
|
|
if token.Error() == nil {
|
|
connecterr = errors.New("connect timeout")
|
|
} else {
|
|
connecterr = token.Error()
|
|
}
|
|
if err := checkBroker(cfg.Server.Server, cfg.Server.MQPort); err != nil {
|
|
logger.Log(0, "could not connect to broker", cfg.Server.Server, err.Error())
|
|
}
|
|
}
|
|
}
|
|
if connecterr != nil {
|
|
logger.Log(0, "failed to establish connection to broker: ", connecterr.Error())
|
|
return connecterr
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// publishes a message to server to update peers on this peer's behalf
|
|
func publishSignal(nodeCfg *config.ClientConfig, signal byte) error {
|
|
if err := publish(nodeCfg, fmt.Sprintf("signal/%s", nodeCfg.Node.ID), []byte{signal}, 1); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func parseNetworkFromTopic(topic string) string {
|
|
return strings.Split(topic, "/")[1]
|
|
}
|
|
|
|
// should only ever use node client configs
|
|
func decryptMsg(nodeCfg *config.ClientConfig, msg []byte) ([]byte, error) {
|
|
if len(msg) <= 24 { // make sure message is of appropriate length
|
|
return nil, fmt.Errorf("recieved invalid message from broker %v", msg)
|
|
}
|
|
|
|
// setup the keys
|
|
diskKey, keyErr := auth.RetrieveTrafficKey(nodeCfg.Node.Network)
|
|
if keyErr != nil {
|
|
return nil, keyErr
|
|
}
|
|
|
|
serverPubKey, err := ncutils.ConvertBytesToKey(nodeCfg.Node.TrafficKeys.Server)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return ncutils.DeChunk(msg, serverPubKey, diskKey)
|
|
}
|
|
|
|
// == Message Caches ==
|
|
|
|
func insert(network, which, cache string) {
|
|
var newMessage = cachedMessage{
|
|
Message: cache,
|
|
LastSeen: time.Now(),
|
|
}
|
|
messageCache.Store(fmt.Sprintf("%s%s", network, which), newMessage)
|
|
}
|
|
|
|
func read(network, which string) string {
|
|
val, isok := messageCache.Load(fmt.Sprintf("%s%s", network, which))
|
|
if isok {
|
|
var readMessage = val.(cachedMessage) // fetch current cached message
|
|
if readMessage.LastSeen.IsZero() {
|
|
return ""
|
|
}
|
|
if time.Now().After(readMessage.LastSeen.Add(time.Hour * 24)) { // check if message has been there over a minute
|
|
messageCache.Delete(fmt.Sprintf("%s%s", network, which)) // remove old message if expired
|
|
return ""
|
|
}
|
|
return readMessage.Message // return current message if not expired
|
|
}
|
|
return ""
|
|
}
|