util.go 2.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. package mq
  2. import (
  3. "errors"
  4. "fmt"
  5. "strings"
  6. "time"
  7. "github.com/gravitl/netmaker/logic"
  8. "github.com/gravitl/netmaker/models"
  9. "github.com/gravitl/netmaker/netclient/ncutils"
  10. )
  11. func decryptMsg(node *models.Node, msg []byte) ([]byte, error) {
  12. if len(msg) <= 24 { // make sure message is of appropriate length
  13. return nil, fmt.Errorf("recieved invalid message from broker %v", msg)
  14. }
  15. trafficKey, trafficErr := logic.RetrievePrivateTrafficKey() // get server private key
  16. if trafficErr != nil {
  17. return nil, trafficErr
  18. }
  19. serverPrivTKey, err := ncutils.ConvertBytesToKey(trafficKey)
  20. if err != nil {
  21. return nil, err
  22. }
  23. nodePubTKey, err := ncutils.ConvertBytesToKey(node.TrafficKeys.Mine)
  24. if err != nil {
  25. return nil, err
  26. }
  27. if strings.Contains(node.Version, "0.10.0") {
  28. return ncutils.BoxDecrypt(msg, nodePubTKey, serverPrivTKey)
  29. }
  30. return ncutils.DeChunk(msg, nodePubTKey, serverPrivTKey)
  31. }
  32. func encryptMsg(node *models.Node, msg []byte) ([]byte, error) {
  33. // fetch server public key to be certain hasn't changed in transit
  34. trafficKey, trafficErr := logic.RetrievePrivateTrafficKey()
  35. if trafficErr != nil {
  36. return nil, trafficErr
  37. }
  38. serverPrivKey, err := ncutils.ConvertBytesToKey(trafficKey)
  39. if err != nil {
  40. return nil, err
  41. }
  42. nodePubKey, err := ncutils.ConvertBytesToKey(node.TrafficKeys.Mine)
  43. if err != nil {
  44. return nil, err
  45. }
  46. if strings.Contains(node.Version, "0.10.0") {
  47. return ncutils.BoxEncrypt(msg, nodePubKey, serverPrivKey)
  48. }
  49. return ncutils.Chunk(msg, nodePubKey, serverPrivKey)
  50. }
  51. func publish(node *models.Node, dest string, msg []byte) error {
  52. encrypted, encryptErr := encryptMsg(node, msg)
  53. if encryptErr != nil {
  54. return encryptErr
  55. }
  56. if mqclient == nil {
  57. return errors.New("cannot publish ... mqclient not connected")
  58. }
  59. if token := mqclient.Publish(dest, 0, true, encrypted); !token.WaitTimeout(MQ_TIMEOUT*time.Second) || token.Error() != nil {
  60. var err error
  61. if token.Error() == nil {
  62. err = errors.New("connection timeout")
  63. } else {
  64. err = token.Error()
  65. }
  66. return err
  67. }
  68. return nil
  69. }
  70. // decodes a message queue topic and returns the embedded node.ID
  71. func getID(topic string) (string, error) {
  72. parts := strings.Split(topic, "/")
  73. count := len(parts)
  74. if count == 1 {
  75. return "", fmt.Errorf("invalid topic")
  76. }
  77. //the last part of the topic will be the node.ID
  78. return parts[count-1], nil
  79. }