mqhandlers.go 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260
  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. if ncutils.IsWindows() {
  104. wireguard.RemoveConfGraceful(nodeCfg.Node.Interface)
  105. }
  106. err = wireguard.ApplyConf(&nodeCfg.Node, nodeCfg.Node.Interface, file)
  107. if err != nil {
  108. ncutils.Log("error restarting wg after node update " + err.Error())
  109. return
  110. }
  111. time.Sleep(time.Second >> 0)
  112. // if newNode.DNSOn == "yes" {
  113. // for _, server := range newNode.NetworkSettings.DefaultServerAddrs {
  114. // if server.IsLeader {
  115. // go local.SetDNSWithRetry(newNode, server.Address)
  116. // break
  117. // }
  118. // }
  119. // }
  120. doneErr := publishSignal(&commsCfg, &nodeCfg, ncutils.DONE)
  121. if doneErr != nil {
  122. ncutils.Log("could not notify server to update peers after interface change")
  123. } else {
  124. ncutils.Log("signalled finished interface update to server")
  125. }
  126. } else if hubChange {
  127. doneErr := publishSignal(&commsCfg, &nodeCfg, ncutils.DONE)
  128. if doneErr != nil {
  129. ncutils.Log("could not notify server to update peers after hub change")
  130. } else {
  131. ncutils.Log("signalled finished hub update to server")
  132. }
  133. }
  134. //deal with DNS
  135. if newNode.DNSOn != "yes" && shouldDNSChange && nodeCfg.Node.Interface != "" {
  136. ncutils.Log("settng DNS off")
  137. if err := removeHostDNS(nodeCfg.Network, ncutils.IsWindows()); err != nil {
  138. ncutils.Log("error removing netmaker profile from /etc/hosts " + err.Error())
  139. }
  140. // _, err := ncutils.RunCmd("/usr/bin/resolvectl revert "+nodeCfg.Node.Interface, true)
  141. // if err != nil {
  142. // ncutils.Log("error applying dns" + err.Error())
  143. // }
  144. }
  145. }
  146. // UpdatePeers -- mqtt message handler for peers/<Network>/<NodeID> topic
  147. func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
  148. var peerUpdate models.PeerUpdate
  149. var network = parseNetworkFromTopic(msg.Topic())
  150. var cfg = config.ClientConfig{}
  151. cfg.Network = network
  152. cfg.ReadConfig()
  153. data, dataErr := decryptMsg(&cfg, msg.Payload())
  154. if dataErr != nil {
  155. return
  156. }
  157. err := json.Unmarshal([]byte(data), &peerUpdate)
  158. if err != nil {
  159. ncutils.Log("error unmarshalling peer data")
  160. return
  161. }
  162. // see if cached hit, if so skip
  163. var currentMessage = read(peerUpdate.Network, lastPeerUpdate)
  164. if currentMessage == string(data) {
  165. return
  166. }
  167. insert(peerUpdate.Network, lastPeerUpdate, string(data))
  168. file := ncutils.GetNetclientPathSpecific() + cfg.Node.Interface + ".conf"
  169. err = wireguard.UpdateWgPeers(file, peerUpdate.Peers)
  170. if err != nil {
  171. ncutils.Log("error updating wireguard peers" + err.Error())
  172. return
  173. }
  174. //err = wireguard.SyncWGQuickConf(cfg.Node.Interface, file)
  175. var iface = cfg.Node.Interface
  176. if ncutils.IsMac() {
  177. iface, err = local.GetMacIface(cfg.Node.Address)
  178. if err != nil {
  179. ncutils.Log("error retrieving mac iface: " + err.Error())
  180. return
  181. }
  182. }
  183. err = wireguard.SetPeers(iface, &cfg.Node, peerUpdate.Peers)
  184. if err != nil {
  185. ncutils.Log("error syncing wg after peer update: " + err.Error())
  186. return
  187. }
  188. ncutils.Log("received peer update for node " + cfg.Node.Name + " " + cfg.Node.Network)
  189. //skip dns updates if this is a peer update for comms network
  190. if cfg.Node.NetworkSettings.IsComms == "yes" {
  191. return
  192. }
  193. if cfg.Node.DNSOn == "yes" {
  194. if err := setHostDNS(peerUpdate.DNS, cfg.Node.Network, ncutils.IsWindows()); err != nil {
  195. ncutils.Log("error updating /etc/hosts " + err.Error())
  196. return
  197. }
  198. } else {
  199. if err := removeHostDNS(cfg.Node.Network, ncutils.IsWindows()); err != nil {
  200. ncutils.Log("error removing profile from /etc/hosts " + err.Error())
  201. return
  202. }
  203. }
  204. }
  205. func setHostDNS(dns, network string, windows bool) error {
  206. etchosts := "/etc/hosts"
  207. if windows {
  208. etchosts = "c:\\windows\\system32\\drivers\\etc\\hosts"
  209. }
  210. dnsdata := strings.NewReader(dns)
  211. profile, err := parser.ParseProfile(dnsdata)
  212. if err != nil {
  213. return err
  214. }
  215. hosts, err := file.NewFile(etchosts)
  216. if err != nil {
  217. return err
  218. }
  219. profile.Name = network
  220. profile.Status = types.Enabled
  221. if err := hosts.ReplaceProfile(profile); err != nil {
  222. return err
  223. }
  224. if err := hosts.Flush(); err != nil {
  225. return err
  226. }
  227. return nil
  228. }
  229. func removeHostDNS(network string, windows bool) error {
  230. etchosts := "/etc/hosts"
  231. if windows {
  232. etchosts = "c:\\windows\\system32\\drivers\\etc\\hosts"
  233. }
  234. hosts, err := file.NewFile(etchosts)
  235. if err != nil {
  236. return err
  237. }
  238. if err := hosts.RemoveProfile(network); err != nil {
  239. return err
  240. }
  241. if err := hosts.Flush(); err != nil {
  242. return err
  243. }
  244. return nil
  245. }