package mq import ( "encoding/json" "errors" "fmt" "net" "time" "github.com/gravitl/netmaker/logger" "github.com/gravitl/netmaker/logic" "github.com/gravitl/netmaker/models" "github.com/gravitl/netmaker/servercfg" "golang.zx2c4.com/wireguard/wgctrl/wgtypes" ) // PublishHostPeerUpdate --- determines and publishes a peer update to one host func PublishHostPeerUpdate(host *models.Host, allNodes []models.Node) error { peerUpdate, err := logic.GetPeerUpdateForHost(host, allNodes) if err != nil { return err } if len(peerUpdate.Peers) == 0 { // no peers to send return nil } data, err := json.Marshal(&peerUpdate) if err != nil { return err } return publish(host, fmt.Sprintf("peers/host/%s/%s", host.ID.String(), servercfg.GetServer()), data) } // FlushNetworkPeersToHost - sends all the peers in the network to the host. func FlushNetworkPeersToHost(client models.Client, networkClients []models.Client) error { logger.Log(0, "flushing network peers to host: ", client.Host.ID.String(), client.Node.Network) addPeerAction := models.PeerAction{ Action: models.AddPeer, Peers: []wgtypes.PeerConfig{}, } rmPeerAction := models.PeerAction{ Action: models.RemovePeer, Peers: []wgtypes.PeerConfig{}, } for _, clientI := range networkClients { clientI := clientI if clientI.Node.ID == client.Node.ID { // skip self continue } allowedIPs := logic.GetAllowedIPs(client, clientI) peerCfg := wgtypes.PeerConfig{ PublicKey: clientI.Host.PublicKey, Endpoint: &net.UDPAddr{ IP: clientI.Host.EndpointIP, Port: logic.GetPeerListenPort(&clientI.Host), }, PersistentKeepaliveInterval: &clientI.Node.PersistentKeepalive, ReplaceAllowedIPs: true, AllowedIPs: allowedIPs, } if len(peerCfg.AllowedIPs) == 0 || (client.Node.IsRelayed && (client.Node.RelayedBy != clientI.Node.ID.String())) { // remove peer if not allowed rmPeerAction.Peers = append(rmPeerAction.Peers, wgtypes.PeerConfig{ PublicKey: clientI.Host.PublicKey, Remove: true, }) continue } addPeerAction.Peers = append(addPeerAction.Peers, peerCfg) } if client.Node.IsIngressGateway { extPeers, _, err := logic.GetExtPeers(&client.Node) if err == nil { addPeerAction.Peers = append(addPeerAction.Peers, extPeers...) } } if len(rmPeerAction.Peers) > 0 { data, err := json.Marshal(rmPeerAction) if err != nil { return err } publish(&client.Host, fmt.Sprintf("peer/host/%s/%s", client.Host.ID.String(), servercfg.GetServer()), data) } if len(addPeerAction.Peers) > 0 { data, err := json.Marshal(addPeerAction) if err != nil { return err } publish(&client.Host, fmt.Sprintf("peer/host/%s/%s", client.Host.ID.String(), servercfg.GetServer()), data) } // send fw update if gw host if client.Node.IsIngressGateway || client.Node.IsEgressGateway { f, err := logic.GetFwUpdate(&client.Host) if err == nil { PublishFwUpdate(&client.Host, &f) } } return nil } // BroadcastDelPeer - notifys all the hosts in the network to remove peer func BroadcastDelPeer(host *models.Host, networkClients []models.Client) error { p := models.PeerAction{ Action: models.RemovePeer, Peers: []wgtypes.PeerConfig{ { PublicKey: host.PublicKey, Endpoint: &net.UDPAddr{ IP: host.EndpointIP, Port: logic.GetPeerListenPort(host), }, ReplaceAllowedIPs: true, UpdateOnly: true, Remove: true, }, }, } data, err := json.Marshal(p) if err != nil { return err } for _, clientI := range networkClients { if clientI.Host.ID == host.ID { // skip self... continue } allowedIPs := logic.GetAllowedIPs(clientI, models.Client{Host: *host}) if len(allowedIPs) != 0 { p.Peers[0].Remove = false p.Peers[0].AllowedIPs = allowedIPs } publish(&clientI.Host, fmt.Sprintf("peer/host/%s/%s", clientI.Host.ID.String(), servercfg.GetServer()), data) if clientI.Node.IsIngressGateway || clientI.Node.IsEgressGateway { go func(peerHost models.Host) { f, err := logic.GetFwUpdate(&peerHost) if err == nil { PublishFwUpdate(&peerHost, &f) } }(clientI.Host) } } return nil } // BroadcastAclUpdate - sends new acl updates to peers func BroadcastAclUpdate(network string) error { clients, err := logic.GetNetworkClients(network) if err != nil { return err } for _, client := range clients { client := client go FlushNetworkPeersToHost(client, clients) } return err } // BroadcastHostUpdate - notifys the hosts in the network to update peer. func BroadcastHostUpdate(host *models.Host, remove bool) error { p := models.PeerAction{ Action: models.UpdatePeer, Peers: []wgtypes.PeerConfig{ { PublicKey: host.PublicKey, Endpoint: &net.UDPAddr{ IP: host.EndpointIP, Port: logic.GetPeerListenPort(host), }, ReplaceAllowedIPs: true, Remove: remove, }, }, } if remove { p.Action = models.RemovePeer } peerHosts := logic.GetRelatedHosts(host.ID.String()) for _, peerHost := range peerHosts { if !remove { p.Peers[0].AllowedIPs = logic.GetAllowedIPs(models.Client{Host: peerHost}, models.Client{Host: *host}) } data, err := json.Marshal(p) if err != nil { return err } publish(&peerHost, fmt.Sprintf("peer/host/%s/%s", peerHost.ID.String(), servercfg.GetServer()), data) } return nil } // BroadcastAddOrUpdateNetworkPeer - notifys the hosts in the network to add or update peer. func BroadcastAddOrUpdateNetworkPeer(client models.Client, update bool) error { clients, err := logic.GetNetworkClients(client.Node.Network) if err != nil { return err } p := models.PeerAction{ Action: models.AddPeer, Peers: []wgtypes.PeerConfig{ { PublicKey: client.Host.PublicKey, Endpoint: &net.UDPAddr{ IP: client.Host.EndpointIP, Port: logic.GetPeerListenPort(&client.Host), }, PersistentKeepaliveInterval: &client.Node.PersistentKeepalive, ReplaceAllowedIPs: true, }, }, } var relayPeerCfg models.PeerAction var relayClient models.Client if client.Node.IsRelayed { relayNode, err := logic.GetNodeByID(client.Node.RelayedBy) if err != nil { return err } relayHost, err := logic.GetHost(relayNode.HostID.String()) if err != nil { return err } relayClient = models.Client{ Host: *relayHost, Node: relayNode, } relayPeerCfg = models.PeerAction{ Action: models.AddPeer, Peers: []wgtypes.PeerConfig{ { PublicKey: relayHost.PublicKey, Endpoint: &net.UDPAddr{ IP: relayHost.EndpointIP, Port: logic.GetPeerListenPort(relayHost), }, PersistentKeepaliveInterval: &relayNode.PersistentKeepalive, ReplaceAllowedIPs: true, }, }, } } if update { p.Action = models.UpdatePeer relayPeerCfg.Action = models.UpdatePeer } for _, clientI := range clients { clientI := clientI if clientI.Node.ID.String() == client.Node.ID.String() { // skip self... continue } // update allowed ips, according to the peer node p.Peers[0].AllowedIPs = logic.GetAllowedIPs(clientI, client) if client.Node.IsRelayed { relayPeerCfg.Peers[0].AllowedIPs = logic.GetAllowedIPs(clientI, relayClient) } if update && len(p.Peers[0].AllowedIPs) == 0 { // remove peer p.Action = models.RemovePeer p.Peers[0].Remove = true } peerHost, err := logic.GetHost(clientI.Host.ID.String()) if err != nil { continue } if clientI.Node.IsRelayed { r := models.PeerAction{ Action: models.AddPeer, } // update the relay peer on this node relayNode, err := logic.GetNodeByID(clientI.Node.RelayedBy) if err != nil { continue } relayHost, err := logic.GetHost(relayNode.HostID.String()) if err != nil { continue } relayedClient := models.Client{ Host: *peerHost, Node: clientI.Node, } relayClient := models.Client{ Host: *relayHost, Node: relayNode, } rPeerCfg := logic.GetPeerConfForRelayed(relayedClient, relayClient) if update { r.Action = models.UpdatePeer } r.Peers = append(r.Peers, rPeerCfg) data, err := json.Marshal(r) if err != nil { continue } publish(peerHost, fmt.Sprintf("peer/host/%s/%s", peerHost.ID.String(), servercfg.GetServer()), data) } else { var data []byte if client.Node.IsRelayed && client.Node.RelayedBy != clientI.Node.ID.String() { data, err = json.Marshal(relayPeerCfg) if err != nil { continue } } else { data, err = json.Marshal(p) if err != nil { continue } } publish(peerHost, fmt.Sprintf("peer/host/%s/%s", peerHost.ID.String(), servercfg.GetServer()), data) } if clientI.Node.IsIngressGateway || clientI.Node.IsEgressGateway { go func(peerHost models.Host) { f, err := logic.GetFwUpdate(&peerHost) if err == nil { PublishFwUpdate(&peerHost, &f) } }(*peerHost) } } return nil } // BroadcastExtClient - publishes msg to add/updates ext client in the network func BroadcastExtClient(ingressClient models.Client) error { clients, err := logic.GetNetworkClients(ingressClient.Node.Network) if err != nil { return err } //flush peers to ingress host go FlushNetworkPeersToHost(ingressClient, clients) // broadcast to update ingress peer to other hosts go BroadcastAddOrUpdateNetworkPeer(ingressClient, true) return nil } // BroadcastDelExtClient - published msg to remove ext client from network func BroadcastDelExtClient(ingressClient models.Client, extclients []models.ExtClient) error { go BroadcastAddOrUpdateNetworkPeer(ingressClient, true) peers := []wgtypes.PeerConfig{} for _, extclient := range extclients { extPubKey, err := wgtypes.ParseKey(extclient.PublicKey) if err != nil { continue } peers = append(peers, wgtypes.PeerConfig{ PublicKey: extPubKey, Remove: true, }) } p := models.PeerAction{ Action: models.RemovePeer, Peers: peers, } data, err := json.Marshal(p) if err != nil { return err } err = publish(&ingressClient.Host, fmt.Sprintf("peer/host/%s/%s", ingressClient.Host.ID.String(), servercfg.GetServer()), data) if err != nil { return err } return nil } // NodeUpdate -- publishes a node update func NodeUpdate(node *models.Node) error { host, err := logic.GetHost(node.HostID.String()) if err != nil { return nil } logger.Log(3, "publishing node update to "+node.ID.String()) //if len(node.NetworkSettings.AccessKeys) > 0 { //node.NetworkSettings.AccessKeys = []models.AccessKey{} // not to be sent (don't need to spread access keys around the network; we need to know how to reach other nodes, not become them) //} data, err := json.Marshal(node) if err != nil { logger.Log(2, "error marshalling node update ", err.Error()) return err } if err = publish(host, fmt.Sprintf("node/update/%s/%s", node.Network, node.ID), data); err != nil { logger.Log(2, "error publishing node update to peer ", node.ID.String(), err.Error()) return err } return nil } // HostUpdate -- publishes a host update to clients func HostUpdate(hostUpdate *models.HostUpdate) error { logger.Log(4, "publishing host update to "+hostUpdate.Host.ID.String()) data, err := json.Marshal(hostUpdate) if err != nil { logger.Log(2, "error marshalling node update ", err.Error()) return err } if err = publish(&hostUpdate.Host, fmt.Sprintf("host/update/%s/%s", hostUpdate.Host.ID.String(), servercfg.GetServer()), data); err != nil { logger.Log(2, "error publishing host update to", hostUpdate.Host.ID.String(), err.Error()) return err } return nil } // ServerStartNotify - notifies all non server nodes to pull changes after a restart func ServerStartNotify() error { nodes, err := logic.GetAllNodes() if err != nil { return err } for i := range nodes { nodes[i].Action = models.NODE_FORCE_UPDATE if err = NodeUpdate(&nodes[i]); err != nil { logger.Log(1, "error when notifying node", nodes[i].ID.String(), "of a server startup") } } return nil } // PublishDNSUpdate publishes a dns update to all nodes on a network func PublishDNSUpdate(network string, dns models.DNSUpdate) error { nodes, err := logic.GetNetworkNodes(network) if err != nil { return err } for _, node := range nodes { host, err := logic.GetHost(node.HostID.String()) if err != nil { logger.Log(0, "error retrieving host for dns update", host.ID.String(), err.Error()) continue } data, err := json.Marshal(dns) if err != nil { logger.Log(0, "failed to encode dns data for node", node.ID.String(), err.Error()) } if err := publish(host, "dns/update/"+host.ID.String()+"/"+servercfg.GetServer(), data); err != nil { logger.Log(0, "error publishing dns update to host", host.ID.String(), err.Error()) continue } logger.Log(3, "published dns update to host", host.ID.String()) } return nil } // PublishAllDNS publishes an array of dns updates (ip / host.network) for each peer to a node joining a network func PublishAllDNS(newnode *models.Node) error { alldns := []models.DNSUpdate{} newnodeHost, err := logic.GetHost(newnode.HostID.String()) if err != nil { return fmt.Errorf("error retrieving host for dns update %w", err) } alldns = append(alldns, getNodeDNS(newnode.Network)...) alldns = append(alldns, getExtClientDNS(newnode.Network)...) alldns = append(alldns, getCustomDNS(newnode.Network)...) data, err := json.Marshal(alldns) if err != nil { return fmt.Errorf("error encoding dns data %w", err) } if err := publish(newnodeHost, "dns/all/"+newnodeHost.ID.String()+"/"+servercfg.GetServer(), data); err != nil { return fmt.Errorf("error publishing full dns update to %s, %w", newnodeHost.ID.String(), err) } logger.Log(3, "published full dns update to %s", newnodeHost.ID.String()) return nil } // PublishDNSDelete publish a dns update deleting a node to all hosts on a network func PublishDNSDelete(node *models.Node, host *models.Host) error { dns := models.DNSUpdate{ Action: models.DNSDeleteByIP, Name: host.Name + "." + node.Network, } if node.Address.IP != nil { dns.Address = node.Address.IP.String() if err := PublishDNSUpdate(node.Network, dns); err != nil { return fmt.Errorf("dns update node deletion %w", err) } } if node.Address6.IP != nil { dns.Address = node.Address6.IP.String() if err := PublishDNSUpdate(node.Network, dns); err != nil { return fmt.Errorf("dns update node deletion %w", err) } } return nil } // PublishReplaceDNS publish a dns update to replace a dns entry on all hosts in network func PublishReplaceDNS(oldNode, newNode *models.Node, host *models.Host) error { dns := models.DNSUpdate{ Action: models.DNSReplaceIP, Name: host.Name + "." + oldNode.Network, } if !oldNode.Address.IP.Equal(newNode.Address.IP) { dns.Address = oldNode.Address.IP.String() dns.NewAddress = newNode.Address.IP.String() if err := PublishDNSUpdate(oldNode.Network, dns); err != nil { return err } } if !oldNode.Address6.IP.Equal(newNode.Address6.IP) { dns.Address = oldNode.Address6.IP.String() dns.NewAddress = newNode.Address6.IP.String() if err := PublishDNSUpdate(oldNode.Network, dns); err != nil { return err } } return nil } // PublishExtClientDNS publish dns update for new extclient func PublishExtCLientDNS(client *models.ExtClient) error { errMsgs := models.DNSError{} dns := models.DNSUpdate{ Action: models.DNSInsert, Name: client.ClientID + "." + client.Network, Address: client.Address, } if client.Address != "" { dns.Address = client.Address if err := PublishDNSUpdate(client.Network, dns); err != nil { errMsgs.ErrorStrings = append(errMsgs.ErrorStrings, err.Error()) } } if client.Address6 != "" { dns.Address = client.Address6 if err := PublishDNSUpdate(client.Network, dns); err != nil { errMsgs.ErrorStrings = append(errMsgs.ErrorStrings, err.Error()) } } if len(errMsgs.ErrorStrings) > 0 { return errMsgs } return nil } // PublishExtClientDNSUpdate update for extclient name change func PublishExtClientDNSUpdate(old, new models.ExtClient, network string) error { dns := models.DNSUpdate{ Action: models.DNSReplaceName, Name: old.ClientID + "." + network, NewName: new.ClientID + "." + network, } if err := PublishDNSUpdate(network, dns); err != nil { return err } return nil } // PublishDeleteExtClientDNS publish dns update to delete extclient entry func PublishDeleteExtClientDNS(client *models.ExtClient) error { dns := models.DNSUpdate{ Action: models.DNSDeleteByName, Name: client.ClientID + "." + client.Network, } if err := PublishDNSUpdate(client.Network, dns); err != nil { return err } return nil } // PublishCustomDNS publish dns update for new custom dns entry func PublishCustomDNS(entry *models.DNSEntry) error { dns := models.DNSUpdate{ Action: models.DNSInsert, Name: entry.Name + "." + entry.Network, //entry.Address6 is never used Address: entry.Address, } if err := PublishDNSUpdate(entry.Network, dns); err != nil { return err } return nil } // PublishHostDNSUpdate publishes dns update on host name change func PublishHostDNSUpdate(old, new *models.Host, networks []string) error { errMsgs := models.DNSError{} for _, network := range networks { dns := models.DNSUpdate{ Action: models.DNSReplaceName, Name: old.Name + "." + network, NewName: new.Name + "." + network, } if err := PublishDNSUpdate(network, dns); err != nil { errMsgs.ErrorStrings = append(errMsgs.ErrorStrings, err.Error()) } } if len(errMsgs.ErrorStrings) > 0 { return errMsgs } return nil } // PublishFwUpdate - publishes fw update to host func PublishFwUpdate(gwHost *models.Host, f *models.FwUpdate) error { data, err := json.Marshal(f) if err != nil { return err } return publish(gwHost, fmt.Sprintf("fw/host/%s/%s", gwHost.ID.String(), servercfg.GetServer()), data) } func pushMetricsToExporter(metrics models.Metrics) error { logger.Log(2, "----> Pushing metrics to exporter") data, err := json.Marshal(metrics) if err != nil { return errors.New("failed to marshal metrics: " + err.Error()) } if token := mqclient.Publish("metrics_exporter", 2, true, data); !token.WaitTimeout(MQ_TIMEOUT*time.Second) || token.Error() != nil { var err error if token.Error() == nil { err = errors.New("connection timeout") } else { err = token.Error() } return err } return nil } func getNodeDNS(network string) []models.DNSUpdate { alldns := []models.DNSUpdate{} dns := models.DNSUpdate{} nodes, err := logic.GetNetworkNodes(network) if err != nil { logger.Log(0, "error retreiving network nodes for network", network, err.Error()) } for _, node := range nodes { host, err := logic.GetHost(node.HostID.String()) if err != nil { logger.Log(0, "error retrieving host for dns update", host.ID.String(), err.Error()) continue } dns.Action = models.DNSInsert dns.Name = host.Name + "." + node.Network if node.Address.IP != nil { dns.Address = node.Address.IP.String() alldns = append(alldns, dns) } if node.Address6.IP != nil { dns.Address = node.Address6.IP.String() alldns = append(alldns, dns) } } return alldns } func getExtClientDNS(network string) []models.DNSUpdate { alldns := []models.DNSUpdate{} dns := models.DNSUpdate{} clients, err := logic.GetNetworkExtClients(network) if err != nil { logger.Log(0, "error retrieving extclients", err.Error()) } for _, client := range clients { dns.Action = models.DNSInsert dns.Name = client.ClientID + "." + client.Network if client.Address != "" { dns.Address = client.Address alldns = append(alldns, dns) } if client.Address6 != "" { dns.Address = client.Address alldns = append(alldns, dns) } } return alldns } func getCustomDNS(network string) []models.DNSUpdate { alldns := []models.DNSUpdate{} dns := models.DNSUpdate{} customdns, err := logic.GetCustomDNS(network) if err != nil { logger.Log(0, "error retrieving custom dns entries", err.Error()) } for _, custom := range customdns { dns.Action = models.DNSInsert dns.Address = custom.Address dns.Name = custom.Name + "." + custom.Network alldns = append(alldns, dns) } return alldns } // sendPeers - retrieve networks, send peer ports to all peers func sendPeers() { hosts, err := logic.GetAllHosts() if err != nil && len(hosts) > 0 { logger.Log(1, "error retrieving networks for keepalive", err.Error()) } var force bool peer_force_send++ if peer_force_send == 5 { servercfg.SetHost() force = true peer_force_send = 0 err := logic.TimerCheckpoint() // run telemetry & log dumps if 24 hours has passed.. if err != nil { logger.Log(3, "error occurred on timer,", err.Error()) } //collectServerMetrics(networks[:]) } if force { allNodes, err := logic.GetAllNodes() if err != nil { return } for _, host := range hosts { host := host logger.Log(2, "sending scheduled peer update (5 min)") if err = PublishHostPeerUpdate(&host, allNodes); err != nil { logger.Log(1, "error publishing peer updates for host: ", host.ID.String(), " Err: ", err.Error()) } } } }