lighthouse.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750
  1. package nebula
  2. import (
  3. "encoding/binary"
  4. "errors"
  5. "fmt"
  6. "net"
  7. "sync"
  8. "time"
  9. "github.com/golang/protobuf/proto"
  10. "github.com/rcrowley/go-metrics"
  11. "github.com/sirupsen/logrus"
  12. )
  13. //TODO: nodes are roaming lighthouses, this is bad. How are they learning?
  14. var ErrHostNotKnown = errors.New("host not known")
  15. // The maximum number of ip addresses to store for a given vpnIp per address family
  16. const maxAddrs = 10
  17. type ip4And6 struct {
  18. //TODO: adding a lock here could allow us to release the lock on lh.addrMap quicker
  19. // v4 and v6 store addresses that have been self reported by the client in a server or where all addresses are stored on a client
  20. v4 []*Ip4AndPort
  21. v6 []*Ip6AndPort
  22. // Learned addresses are ones that a client does not know about but a lighthouse learned from as a result of the received packet
  23. // This is only used if you are a lighthouse server
  24. learnedV4 []*Ip4AndPort
  25. learnedV6 []*Ip6AndPort
  26. }
  27. type LightHouse struct {
  28. //TODO: We need a timer wheel to kick out vpnIps that haven't reported in a long time
  29. sync.RWMutex //Because we concurrently read and write to our maps
  30. amLighthouse bool
  31. myVpnIp uint32
  32. myVpnZeros uint32
  33. punchConn *udpConn
  34. // Local cache of answers from light houses
  35. addrMap map[uint32]*ip4And6
  36. // filters remote addresses allowed for each host
  37. // - When we are a lighthouse, this filters what addresses we store and
  38. // respond with.
  39. // - When we are not a lighthouse, this filters which addresses we accept
  40. // from lighthouses.
  41. remoteAllowList *AllowList
  42. // filters local addresses that we advertise to lighthouses
  43. localAllowList *AllowList
  44. // used to trigger the HandshakeManager when we receive HostQueryReply
  45. handshakeTrigger chan<- uint32
  46. // staticList exists to avoid having a bool in each addrMap entry
  47. // since static should be rare
  48. staticList map[uint32]struct{}
  49. lighthouses map[uint32]struct{}
  50. interval int
  51. nebulaPort uint32 // 32 bits because protobuf does not have a uint16
  52. punchBack bool
  53. punchDelay time.Duration
  54. metrics *MessageMetrics
  55. metricHolepunchTx metrics.Counter
  56. l *logrus.Logger
  57. }
  58. type EncWriter interface {
  59. SendMessageToVpnIp(t NebulaMessageType, st NebulaMessageSubType, vpnIp uint32, p, nb, out []byte)
  60. }
  61. func NewLightHouse(l *logrus.Logger, amLighthouse bool, myVpnIpNet *net.IPNet, ips []uint32, interval int, nebulaPort uint32, pc *udpConn, punchBack bool, punchDelay time.Duration, metricsEnabled bool) *LightHouse {
  62. ones, _ := myVpnIpNet.Mask.Size()
  63. h := LightHouse{
  64. amLighthouse: amLighthouse,
  65. myVpnIp: ip2int(myVpnIpNet.IP),
  66. myVpnZeros: uint32(32 - ones),
  67. addrMap: make(map[uint32]*ip4And6),
  68. nebulaPort: nebulaPort,
  69. lighthouses: make(map[uint32]struct{}),
  70. staticList: make(map[uint32]struct{}),
  71. interval: interval,
  72. punchConn: pc,
  73. punchBack: punchBack,
  74. punchDelay: punchDelay,
  75. l: l,
  76. }
  77. if metricsEnabled {
  78. h.metrics = newLighthouseMetrics()
  79. h.metricHolepunchTx = metrics.GetOrRegisterCounter("messages.tx.holepunch", nil)
  80. } else {
  81. h.metricHolepunchTx = metrics.NilCounter{}
  82. }
  83. for _, ip := range ips {
  84. h.lighthouses[ip] = struct{}{}
  85. }
  86. return &h
  87. }
  88. func (lh *LightHouse) SetRemoteAllowList(allowList *AllowList) {
  89. lh.Lock()
  90. defer lh.Unlock()
  91. lh.remoteAllowList = allowList
  92. }
  93. func (lh *LightHouse) SetLocalAllowList(allowList *AllowList) {
  94. lh.Lock()
  95. defer lh.Unlock()
  96. lh.localAllowList = allowList
  97. }
  98. func (lh *LightHouse) ValidateLHStaticEntries() error {
  99. for lhIP, _ := range lh.lighthouses {
  100. if _, ok := lh.staticList[lhIP]; !ok {
  101. return fmt.Errorf("Lighthouse %s does not have a static_host_map entry", IntIp(lhIP))
  102. }
  103. }
  104. return nil
  105. }
  106. func (lh *LightHouse) Query(ip uint32, f EncWriter) ([]*udpAddr, error) {
  107. //TODO: we need to hold the lock through the next func
  108. if !lh.IsLighthouseIP(ip) {
  109. lh.QueryServer(ip, f)
  110. }
  111. lh.RLock()
  112. if v, ok := lh.addrMap[ip]; ok {
  113. lh.RUnlock()
  114. return TransformLHReplyToUdpAddrs(v), nil
  115. }
  116. lh.RUnlock()
  117. return nil, ErrHostNotKnown
  118. }
  119. // This is asynchronous so no reply should be expected
  120. func (lh *LightHouse) QueryServer(ip uint32, f EncWriter) {
  121. if !lh.amLighthouse {
  122. // Send a query to the lighthouses and hope for the best next time
  123. query, err := proto.Marshal(NewLhQueryByInt(ip))
  124. if err != nil {
  125. lh.l.WithError(err).WithField("vpnIp", IntIp(ip)).Error("Failed to marshal lighthouse query payload")
  126. return
  127. }
  128. lh.metricTx(NebulaMeta_HostQuery, int64(len(lh.lighthouses)))
  129. nb := make([]byte, 12, 12)
  130. out := make([]byte, mtu)
  131. for n := range lh.lighthouses {
  132. f.SendMessageToVpnIp(lightHouse, 0, n, query, nb, out)
  133. }
  134. }
  135. }
  136. func (lh *LightHouse) QueryCache(ip uint32) []*udpAddr {
  137. //TODO: we need to hold the lock through the next func
  138. lh.RLock()
  139. if v, ok := lh.addrMap[ip]; ok {
  140. lh.RUnlock()
  141. return TransformLHReplyToUdpAddrs(v)
  142. }
  143. lh.RUnlock()
  144. return nil
  145. }
  146. //
  147. func (lh *LightHouse) queryAndPrepMessage(ip uint32, f func(*ip4And6) (int, error)) (bool, int, error) {
  148. lh.RLock()
  149. if v, ok := lh.addrMap[ip]; ok {
  150. n, err := f(v)
  151. lh.RUnlock()
  152. return true, n, err
  153. }
  154. lh.RUnlock()
  155. return false, 0, nil
  156. }
  157. func (lh *LightHouse) DeleteVpnIP(vpnIP uint32) {
  158. // First we check the static mapping
  159. // and do nothing if it is there
  160. if _, ok := lh.staticList[vpnIP]; ok {
  161. return
  162. }
  163. lh.Lock()
  164. //l.Debugln(lh.addrMap)
  165. delete(lh.addrMap, vpnIP)
  166. if lh.l.Level >= logrus.DebugLevel {
  167. lh.l.Debugf("deleting %s from lighthouse.", IntIp(vpnIP))
  168. }
  169. lh.Unlock()
  170. }
  171. // AddRemote is correct way for non LightHouse members to add an address. toAddr will be placed in the learned map
  172. // static means this is a static host entry from the config file, it should only be used on start up
  173. func (lh *LightHouse) AddRemote(vpnIP uint32, toAddr *udpAddr, static bool) {
  174. if ipv4 := toAddr.IP.To4(); ipv4 != nil {
  175. lh.addRemoteV4(vpnIP, NewIp4AndPort(ipv4, uint32(toAddr.Port)), static, true)
  176. } else {
  177. lh.addRemoteV6(vpnIP, NewIp6AndPort(toAddr.IP, uint32(toAddr.Port)), static, true)
  178. }
  179. //TODO: if we do not add due to a config filter we may end up not having any addresses here
  180. if static {
  181. lh.staticList[vpnIP] = struct{}{}
  182. }
  183. }
  184. // unlockedGetAddrs assumes you have the lh lock
  185. func (lh *LightHouse) unlockedGetAddrs(vpnIP uint32) *ip4And6 {
  186. am, ok := lh.addrMap[vpnIP]
  187. if !ok {
  188. am = &ip4And6{}
  189. lh.addrMap[vpnIP] = am
  190. }
  191. return am
  192. }
  193. // addRemoteV4 is a lighthouse internal method that prepends a remote if it is allowed by the allow list and not duplicated
  194. func (lh *LightHouse) addRemoteV4(vpnIP uint32, to *Ip4AndPort, static bool, learned bool) {
  195. // First we check if the sender thinks this is a static entry
  196. // and do nothing if it is not, but should be considered static
  197. if static == false {
  198. if _, ok := lh.staticList[vpnIP]; ok {
  199. return
  200. }
  201. }
  202. lh.Lock()
  203. defer lh.Unlock()
  204. am := lh.unlockedGetAddrs(vpnIP)
  205. if learned {
  206. if !lh.unlockedShouldAddV4(am.learnedV4, to) {
  207. return
  208. }
  209. am.learnedV4 = prependAndLimitV4(am.learnedV4, to)
  210. } else {
  211. if !lh.unlockedShouldAddV4(am.v4, to) {
  212. return
  213. }
  214. am.v4 = prependAndLimitV4(am.v4, to)
  215. }
  216. }
  217. func prependAndLimitV4(cache []*Ip4AndPort, to *Ip4AndPort) []*Ip4AndPort {
  218. cache = append(cache, nil)
  219. copy(cache[1:], cache)
  220. cache[0] = to
  221. if len(cache) > MaxRemotes {
  222. cache = cache[:maxAddrs]
  223. }
  224. return cache
  225. }
  226. // unlockedShouldAddV4 checks if to is allowed by our allow list and is not already present in the cache
  227. func (lh *LightHouse) unlockedShouldAddV4(am []*Ip4AndPort, to *Ip4AndPort) bool {
  228. allow := lh.remoteAllowList.AllowIpV4(to.Ip)
  229. if lh.l.Level >= logrus.TraceLevel {
  230. lh.l.WithField("remoteIp", IntIp(to.Ip)).WithField("allow", allow).Trace("remoteAllowList.Allow")
  231. }
  232. if !allow || ipMaskContains(lh.myVpnIp, lh.myVpnZeros, to.Ip) {
  233. return false
  234. }
  235. for _, v := range am {
  236. if v.Ip == to.Ip && v.Port == to.Port {
  237. return false
  238. }
  239. }
  240. return true
  241. }
  242. // addRemoteV6 is a lighthouse internal method that prepends a remote if it is allowed by the allow list and not duplicated
  243. func (lh *LightHouse) addRemoteV6(vpnIP uint32, to *Ip6AndPort, static bool, learned bool) {
  244. // First we check if the sender thinks this is a static entry
  245. // and do nothing if it is not, but should be considered static
  246. if static == false {
  247. if _, ok := lh.staticList[vpnIP]; ok {
  248. return
  249. }
  250. }
  251. lh.Lock()
  252. defer lh.Unlock()
  253. am := lh.unlockedGetAddrs(vpnIP)
  254. if learned {
  255. if !lh.unlockedShouldAddV6(am.learnedV6, to) {
  256. return
  257. }
  258. am.learnedV6 = prependAndLimitV6(am.learnedV6, to)
  259. } else {
  260. if !lh.unlockedShouldAddV6(am.v6, to) {
  261. return
  262. }
  263. am.v6 = prependAndLimitV6(am.v6, to)
  264. }
  265. }
  266. func prependAndLimitV6(cache []*Ip6AndPort, to *Ip6AndPort) []*Ip6AndPort {
  267. cache = append(cache, nil)
  268. copy(cache[1:], cache)
  269. cache[0] = to
  270. if len(cache) > MaxRemotes {
  271. cache = cache[:maxAddrs]
  272. }
  273. return cache
  274. }
  275. // unlockedShouldAddV6 checks if to is allowed by our allow list and is not already present in the cache
  276. func (lh *LightHouse) unlockedShouldAddV6(am []*Ip6AndPort, to *Ip6AndPort) bool {
  277. allow := lh.remoteAllowList.AllowIpV6(to.Hi, to.Lo)
  278. if lh.l.Level >= logrus.TraceLevel {
  279. lh.l.WithField("remoteIp", lhIp6ToIp(to)).WithField("allow", allow).Trace("remoteAllowList.Allow")
  280. }
  281. if !allow {
  282. return false
  283. }
  284. for _, v := range am {
  285. if v.Hi == to.Hi && v.Lo == to.Lo && v.Port == to.Port {
  286. return false
  287. }
  288. }
  289. return true
  290. }
  291. func lhIp6ToIp(v *Ip6AndPort) net.IP {
  292. ip := make(net.IP, 16)
  293. binary.BigEndian.PutUint64(ip[:8], v.Hi)
  294. binary.BigEndian.PutUint64(ip[8:], v.Lo)
  295. return ip
  296. }
  297. func (lh *LightHouse) AddRemoteAndReset(vpnIP uint32, toIp *udpAddr) {
  298. if lh.amLighthouse {
  299. lh.DeleteVpnIP(vpnIP)
  300. lh.AddRemote(vpnIP, toIp, false)
  301. }
  302. }
  303. func (lh *LightHouse) IsLighthouseIP(vpnIP uint32) bool {
  304. if _, ok := lh.lighthouses[vpnIP]; ok {
  305. return true
  306. }
  307. return false
  308. }
  309. func NewLhQueryByInt(VpnIp uint32) *NebulaMeta {
  310. return &NebulaMeta{
  311. Type: NebulaMeta_HostQuery,
  312. Details: &NebulaMetaDetails{
  313. VpnIp: VpnIp,
  314. },
  315. }
  316. }
  317. func NewIp4AndPort(ip net.IP, port uint32) *Ip4AndPort {
  318. ipp := Ip4AndPort{Port: port}
  319. ipp.Ip = ip2int(ip)
  320. return &ipp
  321. }
  322. func NewIp6AndPort(ip net.IP, port uint32) *Ip6AndPort {
  323. return &Ip6AndPort{
  324. Hi: binary.BigEndian.Uint64(ip[:8]),
  325. Lo: binary.BigEndian.Uint64(ip[8:]),
  326. Port: port,
  327. }
  328. }
  329. func NewUDPAddrFromLH4(ipp *Ip4AndPort) *udpAddr {
  330. ip := ipp.Ip
  331. return NewUDPAddr(
  332. net.IPv4(byte(ip&0xff000000>>24), byte(ip&0x00ff0000>>16), byte(ip&0x0000ff00>>8), byte(ip&0x000000ff)),
  333. uint16(ipp.Port),
  334. )
  335. }
  336. func NewUDPAddrFromLH6(ipp *Ip6AndPort) *udpAddr {
  337. return NewUDPAddr(lhIp6ToIp(ipp), uint16(ipp.Port))
  338. }
  339. func (lh *LightHouse) LhUpdateWorker(f EncWriter) {
  340. if lh.amLighthouse || lh.interval == 0 {
  341. return
  342. }
  343. for {
  344. lh.SendUpdate(f)
  345. time.Sleep(time.Second * time.Duration(lh.interval))
  346. }
  347. }
  348. func (lh *LightHouse) SendUpdate(f EncWriter) {
  349. var v4 []*Ip4AndPort
  350. var v6 []*Ip6AndPort
  351. for _, e := range *localIps(lh.l, lh.localAllowList) {
  352. if ip4 := e.To4(); ip4 != nil && ipMaskContains(lh.myVpnIp, lh.myVpnZeros, ip2int(ip4)) {
  353. continue
  354. }
  355. // Only add IPs that aren't my VPN/tun IP
  356. if ip := e.To4(); ip != nil {
  357. v4 = append(v4, NewIp4AndPort(e, lh.nebulaPort))
  358. } else {
  359. v6 = append(v6, NewIp6AndPort(e, lh.nebulaPort))
  360. }
  361. }
  362. m := &NebulaMeta{
  363. Type: NebulaMeta_HostUpdateNotification,
  364. Details: &NebulaMetaDetails{
  365. VpnIp: lh.myVpnIp,
  366. Ip4AndPorts: v4,
  367. Ip6AndPorts: v6,
  368. },
  369. }
  370. lh.metricTx(NebulaMeta_HostUpdateNotification, int64(len(lh.lighthouses)))
  371. nb := make([]byte, 12, 12)
  372. out := make([]byte, mtu)
  373. mm, err := proto.Marshal(m)
  374. if err != nil && lh.l.Level >= logrus.DebugLevel {
  375. lh.l.WithError(err).Error("Error while marshaling for lighthouse update")
  376. return
  377. }
  378. for vpnIp := range lh.lighthouses {
  379. f.SendMessageToVpnIp(lightHouse, 0, vpnIp, mm, nb, out)
  380. }
  381. }
  382. type LightHouseHandler struct {
  383. lh *LightHouse
  384. nb []byte
  385. out []byte
  386. pb []byte
  387. meta *NebulaMeta
  388. l *logrus.Logger
  389. }
  390. func (lh *LightHouse) NewRequestHandler() *LightHouseHandler {
  391. lhh := &LightHouseHandler{
  392. lh: lh,
  393. nb: make([]byte, 12, 12),
  394. out: make([]byte, mtu),
  395. l: lh.l,
  396. pb: make([]byte, mtu),
  397. meta: &NebulaMeta{
  398. Details: &NebulaMetaDetails{},
  399. },
  400. }
  401. return lhh
  402. }
  403. func (lh *LightHouse) metricRx(t NebulaMeta_MessageType, i int64) {
  404. lh.metrics.Rx(NebulaMessageType(t), 0, i)
  405. }
  406. func (lh *LightHouse) metricTx(t NebulaMeta_MessageType, i int64) {
  407. lh.metrics.Tx(NebulaMessageType(t), 0, i)
  408. }
  409. // This method is similar to Reset(), but it re-uses the pointer structs
  410. // so that we don't have to re-allocate them
  411. func (lhh *LightHouseHandler) resetMeta() *NebulaMeta {
  412. details := lhh.meta.Details
  413. lhh.meta.Reset()
  414. // Keep the array memory around
  415. details.Ip4AndPorts = details.Ip4AndPorts[:0]
  416. details.Ip6AndPorts = details.Ip6AndPorts[:0]
  417. lhh.meta.Details = details
  418. return lhh.meta
  419. }
  420. //TODO: do we need c here?
  421. func (lhh *LightHouseHandler) HandleRequest(rAddr *udpAddr, vpnIp uint32, p []byte, w EncWriter) {
  422. n := lhh.resetMeta()
  423. err := n.Unmarshal(p)
  424. if err != nil {
  425. lhh.l.WithError(err).WithField("vpnIp", IntIp(vpnIp)).WithField("udpAddr", rAddr).
  426. Error("Failed to unmarshal lighthouse packet")
  427. //TODO: send recv_error?
  428. return
  429. }
  430. if n.Details == nil {
  431. lhh.l.WithField("vpnIp", IntIp(vpnIp)).WithField("udpAddr", rAddr).
  432. Error("Invalid lighthouse update")
  433. //TODO: send recv_error?
  434. return
  435. }
  436. lhh.lh.metricRx(n.Type, 1)
  437. switch n.Type {
  438. case NebulaMeta_HostQuery:
  439. lhh.handleHostQuery(n, vpnIp, rAddr, w)
  440. case NebulaMeta_HostQueryReply:
  441. lhh.handleHostQueryReply(n, vpnIp)
  442. case NebulaMeta_HostUpdateNotification:
  443. lhh.handleHostUpdateNotification(n, vpnIp)
  444. case NebulaMeta_HostMovedNotification:
  445. case NebulaMeta_HostPunchNotification:
  446. lhh.handleHostPunchNotification(n, vpnIp, w)
  447. }
  448. }
  449. func (lhh *LightHouseHandler) handleHostQuery(n *NebulaMeta, vpnIp uint32, addr *udpAddr, w EncWriter) {
  450. // Exit if we don't answer queries
  451. if !lhh.lh.amLighthouse {
  452. if lhh.l.Level >= logrus.DebugLevel {
  453. lhh.l.Debugln("I don't answer queries, but received from: ", addr)
  454. }
  455. return
  456. }
  457. //TODO: we can DRY this further
  458. reqVpnIP := n.Details.VpnIp
  459. //TODO: Maybe instead of marshalling into n we marshal into a new `r` to not nuke our current request data
  460. //TODO: If we use a lock on cache we can avoid holding it on lh.addrMap and keep things moving better
  461. found, ln, err := lhh.lh.queryAndPrepMessage(n.Details.VpnIp, func(cache *ip4And6) (int, error) {
  462. n = lhh.resetMeta()
  463. n.Type = NebulaMeta_HostQueryReply
  464. n.Details.VpnIp = reqVpnIP
  465. lhh.coalesceAnswers(cache, n)
  466. return n.MarshalTo(lhh.pb)
  467. })
  468. if !found {
  469. return
  470. }
  471. if err != nil {
  472. lhh.l.WithError(err).WithField("vpnIp", IntIp(vpnIp)).Error("Failed to marshal lighthouse host query reply")
  473. return
  474. }
  475. lhh.lh.metricTx(NebulaMeta_HostQueryReply, 1)
  476. w.SendMessageToVpnIp(lightHouse, 0, vpnIp, lhh.pb[:ln], lhh.nb, lhh.out[:0])
  477. // This signals the other side to punch some zero byte udp packets
  478. found, ln, err = lhh.lh.queryAndPrepMessage(vpnIp, func(cache *ip4And6) (int, error) {
  479. n = lhh.resetMeta()
  480. n.Type = NebulaMeta_HostPunchNotification
  481. n.Details.VpnIp = vpnIp
  482. lhh.coalesceAnswers(cache, n)
  483. return n.MarshalTo(lhh.pb)
  484. })
  485. if !found {
  486. return
  487. }
  488. if err != nil {
  489. lhh.l.WithError(err).WithField("vpnIp", IntIp(vpnIp)).Error("Failed to marshal lighthouse host was queried for")
  490. return
  491. }
  492. lhh.lh.metricTx(NebulaMeta_HostPunchNotification, 1)
  493. w.SendMessageToVpnIp(lightHouse, 0, reqVpnIP, lhh.pb[:ln], lhh.nb, lhh.out[:0])
  494. }
  495. func (lhh *LightHouseHandler) coalesceAnswers(cache *ip4And6, n *NebulaMeta) {
  496. n.Details.Ip4AndPorts = append(n.Details.Ip4AndPorts, cache.v4...)
  497. n.Details.Ip4AndPorts = append(n.Details.Ip4AndPorts, cache.learnedV4...)
  498. n.Details.Ip6AndPorts = append(n.Details.Ip6AndPorts, cache.v6...)
  499. n.Details.Ip6AndPorts = append(n.Details.Ip6AndPorts, cache.learnedV6...)
  500. }
  501. func (lhh *LightHouseHandler) handleHostQueryReply(n *NebulaMeta, vpnIp uint32) {
  502. if !lhh.lh.IsLighthouseIP(vpnIp) {
  503. return
  504. }
  505. // We can't just slam the responses in as they may come from multiple lighthouses and we should coalesce the answers
  506. for _, to := range n.Details.Ip4AndPorts {
  507. lhh.lh.addRemoteV4(n.Details.VpnIp, to, false, false)
  508. }
  509. for _, to := range n.Details.Ip6AndPorts {
  510. lhh.lh.addRemoteV6(n.Details.VpnIp, to, false, false)
  511. }
  512. // Non-blocking attempt to trigger, skip if it would block
  513. select {
  514. case lhh.lh.handshakeTrigger <- n.Details.VpnIp:
  515. default:
  516. }
  517. }
  518. func (lhh *LightHouseHandler) handleHostUpdateNotification(n *NebulaMeta, vpnIp uint32) {
  519. if !lhh.lh.amLighthouse {
  520. if lhh.l.Level >= logrus.DebugLevel {
  521. lhh.l.Debugln("I am not a lighthouse, do not take host updates: ", vpnIp)
  522. }
  523. return
  524. }
  525. //Simple check that the host sent this not someone else
  526. if n.Details.VpnIp != vpnIp {
  527. if lhh.l.Level >= logrus.DebugLevel {
  528. lhh.l.WithField("vpnIp", IntIp(vpnIp)).WithField("answer", IntIp(n.Details.VpnIp)).Debugln("Host sent invalid update")
  529. }
  530. return
  531. }
  532. lhh.lh.Lock()
  533. defer lhh.lh.Unlock()
  534. am := lhh.lh.unlockedGetAddrs(vpnIp)
  535. //TODO: other note on a lock for am so we can release more quickly and lock our real unit of change which is far less contended
  536. // We don't accumulate addresses being told to us
  537. am.v4 = am.v4[:0]
  538. am.v6 = am.v6[:0]
  539. for _, v := range n.Details.Ip4AndPorts {
  540. if lhh.lh.unlockedShouldAddV4(am.v4, v) {
  541. am.v4 = append(am.v4, v)
  542. }
  543. }
  544. for _, v := range n.Details.Ip6AndPorts {
  545. if lhh.lh.unlockedShouldAddV6(am.v6, v) {
  546. am.v6 = append(am.v6, v)
  547. }
  548. }
  549. // We prefer the first n addresses if we got too big
  550. if len(am.v4) > MaxRemotes {
  551. am.v4 = am.v4[:MaxRemotes]
  552. }
  553. if len(am.v6) > MaxRemotes {
  554. am.v6 = am.v6[:MaxRemotes]
  555. }
  556. }
  557. func (lhh *LightHouseHandler) handleHostPunchNotification(n *NebulaMeta, vpnIp uint32, w EncWriter) {
  558. if !lhh.lh.IsLighthouseIP(vpnIp) {
  559. return
  560. }
  561. empty := []byte{0}
  562. punch := func(vpnPeer *udpAddr) {
  563. if vpnPeer == nil {
  564. return
  565. }
  566. go func() {
  567. time.Sleep(lhh.lh.punchDelay)
  568. lhh.lh.metricHolepunchTx.Inc(1)
  569. lhh.lh.punchConn.WriteTo(empty, vpnPeer)
  570. }()
  571. if lhh.l.Level >= logrus.DebugLevel {
  572. //TODO: lacking the ip we are actually punching on, old: l.Debugf("Punching %s on %d for %s", IntIp(a.Ip), a.Port, IntIp(n.Details.VpnIp))
  573. lhh.l.Debugf("Punching on %d for %s", vpnPeer.Port, IntIp(n.Details.VpnIp))
  574. }
  575. }
  576. for _, a := range n.Details.Ip4AndPorts {
  577. punch(NewUDPAddrFromLH4(a))
  578. }
  579. for _, a := range n.Details.Ip6AndPorts {
  580. punch(NewUDPAddrFromLH6(a))
  581. }
  582. // This sends a nebula test packet to the host trying to contact us. In the case
  583. // of a double nat or other difficult scenario, this may help establish
  584. // a tunnel.
  585. if lhh.lh.punchBack {
  586. go func() {
  587. time.Sleep(time.Second * 5)
  588. if lhh.l.Level >= logrus.DebugLevel {
  589. lhh.l.Debugf("Sending a nebula test packet to vpn ip %s", IntIp(n.Details.VpnIp))
  590. }
  591. //NOTE: we have to allocate a new output buffer here since we are spawning a new goroutine
  592. // for each punchBack packet. We should move this into a timerwheel or a single goroutine
  593. // managed by a channel.
  594. w.SendMessageToVpnIp(test, testRequest, n.Details.VpnIp, []byte(""), make([]byte, 12, 12), make([]byte, mtu))
  595. }()
  596. }
  597. }
  598. func TransformLHReplyToUdpAddrs(ips *ip4And6) []*udpAddr {
  599. addrs := make([]*udpAddr, len(ips.v4)+len(ips.v6)+len(ips.learnedV4)+len(ips.learnedV6))
  600. i := 0
  601. for _, v := range ips.learnedV4 {
  602. addrs[i] = NewUDPAddrFromLH4(v)
  603. i++
  604. }
  605. for _, v := range ips.v4 {
  606. addrs[i] = NewUDPAddrFromLH4(v)
  607. i++
  608. }
  609. for _, v := range ips.learnedV6 {
  610. addrs[i] = NewUDPAddrFromLH6(v)
  611. i++
  612. }
  613. for _, v := range ips.v6 {
  614. addrs[i] = NewUDPAddrFromLH6(v)
  615. i++
  616. }
  617. return addrs
  618. }
  619. // ipMaskContains checks if testIp is contained by ip after applying a cidr
  620. // zeros is 32 - bits from net.IPMask.Size()
  621. func ipMaskContains(ip uint32, zeros uint32, testIp uint32) bool {
  622. return (testIp^ip)>>zeros == 0
  623. }