mqhandlers.go 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257
  1. package functions
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "runtime"
  6. "strings"
  7. "time"
  8. mqtt "github.com/eclipse/paho.mqtt.golang"
  9. "github.com/gravitl/netmaker/models"
  10. "github.com/gravitl/netmaker/netclient/config"
  11. "github.com/gravitl/netmaker/netclient/local"
  12. "github.com/gravitl/netmaker/netclient/ncutils"
  13. "github.com/gravitl/netmaker/netclient/wireguard"
  14. "github.com/guumaster/hostctl/pkg/file"
  15. "github.com/guumaster/hostctl/pkg/parser"
  16. "github.com/guumaster/hostctl/pkg/types"
  17. "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
  18. )
  19. // All -- mqtt message hander for all ('#') topics
  20. var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
  21. ncutils.Log("default message handler -- received message but not handling")
  22. ncutils.Log("Topic: " + string(msg.Topic()))
  23. //ncutils.Log("Message: " + string(msg.Payload()))
  24. }
  25. // NodeUpdate -- mqtt message handler for /update/<NodeID> topic
  26. func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
  27. var newNode models.Node
  28. var nodeCfg config.ClientConfig
  29. var network = parseNetworkFromTopic(msg.Topic())
  30. nodeCfg.Network = network
  31. nodeCfg.ReadConfig()
  32. var commsCfg = getCommsCfgByNode(&nodeCfg.Node)
  33. data, dataErr := decryptMsg(&nodeCfg, msg.Payload())
  34. if dataErr != nil {
  35. return
  36. }
  37. err := json.Unmarshal([]byte(data), &newNode)
  38. if err != nil {
  39. ncutils.Log("error unmarshalling node update data" + err.Error())
  40. return
  41. }
  42. ncutils.Log("received message to update node " + newNode.Name)
  43. // see if cache hit, if so skip
  44. var currentMessage = read(newNode.Network, lastNodeUpdate)
  45. if currentMessage == string(data) {
  46. return
  47. }
  48. insert(newNode.Network, lastNodeUpdate, string(data)) // store new message in cache
  49. // ensure that OS never changes
  50. newNode.OS = runtime.GOOS
  51. // check if interface needs to delta
  52. ifaceDelta := ncutils.IfaceDelta(&nodeCfg.Node, &newNode)
  53. shouldDNSChange := nodeCfg.Node.DNSOn != newNode.DNSOn
  54. hubChange := nodeCfg.Node.IsHub != newNode.IsHub
  55. nodeCfg.Node = newNode
  56. switch newNode.Action {
  57. case models.NODE_DELETE:
  58. ncutils.PrintLog(fmt.Sprintf("received delete request for %s", nodeCfg.Node.Name), 0)
  59. unsubscribeNode(client, &nodeCfg)
  60. if err = LeaveNetwork(nodeCfg.Node.Network, true); err != nil {
  61. if !strings.Contains("rpc error", err.Error()) {
  62. ncutils.PrintLog(fmt.Sprintf("failed to leave, please check that local files for network %s were removed", nodeCfg.Node.Network), 0)
  63. return
  64. }
  65. }
  66. ncutils.PrintLog(fmt.Sprintf("%s was removed", nodeCfg.Node.Name), 0)
  67. return
  68. case models.NODE_UPDATE_KEY:
  69. // == get the current key for node ==
  70. oldPrivateKey, retErr := wireguard.RetrievePrivKey(nodeCfg.Network)
  71. if retErr != nil {
  72. break
  73. }
  74. if err := UpdateKeys(&nodeCfg, client); err != nil {
  75. ncutils.PrintLog("err updating wireguard keys, reusing last key\n"+err.Error(), 0)
  76. if key, parseErr := wgtypes.ParseKey(oldPrivateKey); parseErr == nil {
  77. wireguard.StorePrivKey(key.String(), nodeCfg.Network)
  78. nodeCfg.Node.PublicKey = key.PublicKey().String()
  79. }
  80. }
  81. ifaceDelta = true
  82. case models.NODE_NOOP:
  83. default:
  84. }
  85. // Save new config
  86. nodeCfg.Node.Action = models.NODE_NOOP
  87. if err := config.Write(&nodeCfg, nodeCfg.Network); err != nil {
  88. ncutils.PrintLog("error updating node configuration: "+err.Error(), 0)
  89. }
  90. nameserver := nodeCfg.Server.CoreDNSAddr
  91. privateKey, err := wireguard.RetrievePrivKey(newNode.Network)
  92. if err != nil {
  93. ncutils.Log("error reading PrivateKey " + err.Error())
  94. return
  95. }
  96. file := ncutils.GetNetclientPathSpecific() + nodeCfg.Node.Interface + ".conf"
  97. if err := wireguard.UpdateWgInterface(file, privateKey, nameserver, newNode); err != nil {
  98. ncutils.Log("error updating wireguard config " + err.Error())
  99. return
  100. }
  101. if ifaceDelta { // if a change caused an ifacedelta we need to notify the server to update the peers
  102. ncutils.Log("applying WG conf to " + file)
  103. err = wireguard.ApplyConf(&nodeCfg.Node, nodeCfg.Node.Interface, file)
  104. if err != nil {
  105. ncutils.Log("error restarting wg after node update " + err.Error())
  106. return
  107. }
  108. time.Sleep(time.Second >> 0)
  109. // if newNode.DNSOn == "yes" {
  110. // for _, server := range newNode.NetworkSettings.DefaultServerAddrs {
  111. // if server.IsLeader {
  112. // go local.SetDNSWithRetry(newNode, server.Address)
  113. // break
  114. // }
  115. // }
  116. // }
  117. doneErr := publishSignal(&commsCfg, &nodeCfg, ncutils.DONE)
  118. if doneErr != nil {
  119. ncutils.Log("could not notify server to update peers after interface change")
  120. } else {
  121. ncutils.Log("signalled finished interface update to server")
  122. }
  123. } else if hubChange {
  124. doneErr := publishSignal(&commsCfg, &nodeCfg, ncutils.DONE)
  125. if doneErr != nil {
  126. ncutils.Log("could not notify server to update peers after hub change")
  127. } else {
  128. ncutils.Log("signalled finished hub update to server")
  129. }
  130. }
  131. //deal with DNS
  132. if newNode.DNSOn != "yes" && shouldDNSChange && nodeCfg.Node.Interface != "" {
  133. ncutils.Log("settng DNS off")
  134. if err := removeHostDNS(nodeCfg.Network, ncutils.IsWindows()); err != nil {
  135. ncutils.Log("error removing netmaker profile from /etc/hosts " + err.Error())
  136. }
  137. // _, err := ncutils.RunCmd("/usr/bin/resolvectl revert "+nodeCfg.Node.Interface, true)
  138. // if err != nil {
  139. // ncutils.Log("error applying dns" + err.Error())
  140. // }
  141. }
  142. }
  143. // UpdatePeers -- mqtt message handler for peers/<Network>/<NodeID> topic
  144. func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
  145. var peerUpdate models.PeerUpdate
  146. var network = parseNetworkFromTopic(msg.Topic())
  147. var cfg = config.ClientConfig{}
  148. cfg.Network = network
  149. cfg.ReadConfig()
  150. data, dataErr := decryptMsg(&cfg, msg.Payload())
  151. if dataErr != nil {
  152. return
  153. }
  154. err := json.Unmarshal([]byte(data), &peerUpdate)
  155. if err != nil {
  156. ncutils.Log("error unmarshalling peer data")
  157. return
  158. }
  159. // see if cached hit, if so skip
  160. var currentMessage = read(peerUpdate.Network, lastPeerUpdate)
  161. if currentMessage == string(data) {
  162. return
  163. }
  164. insert(peerUpdate.Network, lastPeerUpdate, string(data))
  165. file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
  166. err = wireguard.UpdateWgPeers(file, peerUpdate.Peers)
  167. if err != nil {
  168. ncutils.Log("error updating wireguard peers" + err.Error())
  169. return
  170. }
  171. //err = wireguard.SyncWGQuickConf(cfg.Node.Interface, file)
  172. var iface = cfg.Node.Interface
  173. if ncutils.IsMac() {
  174. iface, err = local.GetMacIface(cfg.Node.Address)
  175. if err != nil {
  176. ncutils.Log("error retrieving mac iface: " + err.Error())
  177. return
  178. }
  179. }
  180. err = wireguard.SetPeers(iface, &cfg.Node, peerUpdate.Peers)
  181. if err != nil {
  182. ncutils.Log("error syncing wg after peer update: " + err.Error())
  183. return
  184. }
  185. ncutils.Log("received peer update for node " + cfg.Node.Name + " " + cfg.Node.Network)
  186. //skip dns updates if this is a peer update for comms network
  187. if cfg.Node.NetworkSettings.IsComms == "yes" {
  188. return
  189. }
  190. if cfg.Node.DNSOn == "yes" {
  191. if err := setHostDNS(peerUpdate.DNS, cfg.Node.Network, ncutils.IsWindows()); err != nil {
  192. ncutils.Log("error updating /etc/hosts " + err.Error())
  193. return
  194. }
  195. } else {
  196. if err := removeHostDNS(cfg.Node.Network, ncutils.IsWindows()); err != nil {
  197. ncutils.Log("error removing profile from /etc/hosts " + err.Error())
  198. return
  199. }
  200. }
  201. }
  202. func setHostDNS(dns, network string, windows bool) error {
  203. etchosts := "/etc/hosts"
  204. if windows {
  205. etchosts = "c:\\windows\\system32\\drivers\\etc\\hosts"
  206. }
  207. dnsdata := strings.NewReader(dns)
  208. profile, err := parser.ParseProfile(dnsdata)
  209. if err != nil {
  210. return err
  211. }
  212. hosts, err := file.NewFile(etchosts)
  213. if err != nil {
  214. return err
  215. }
  216. profile.Name = network
  217. profile.Status = types.Enabled
  218. if err := hosts.ReplaceProfile(profile); err != nil {
  219. return err
  220. }
  221. if err := hosts.Flush(); err != nil {
  222. return err
  223. }
  224. return nil
  225. }
  226. func removeHostDNS(network string, windows bool) error {
  227. etchosts := "/etc/hosts"
  228. if windows {
  229. etchosts = "c:\\windows\\system32\\drivers\\etc\\hosts"
  230. }
  231. hosts, err := file.NewFile(etchosts)
  232. if err != nil {
  233. return err
  234. }
  235. if err := hosts.RemoveProfile(network); err != nil {
  236. return err
  237. }
  238. if err := hosts.Flush(); err != nil {
  239. return err
  240. }
  241. return nil
  242. }