handshake_manager.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732
  1. package nebula
  2. import (
  3. "bytes"
  4. "context"
  5. "crypto/rand"
  6. "encoding/binary"
  7. "errors"
  8. "net/netip"
  9. "slices"
  10. "sync"
  11. "time"
  12. "github.com/rcrowley/go-metrics"
  13. "github.com/sirupsen/logrus"
  14. "github.com/slackhq/nebula/cert"
  15. "github.com/slackhq/nebula/header"
  16. "github.com/slackhq/nebula/udp"
  17. )
  18. const (
  19. DefaultHandshakeTryInterval = time.Millisecond * 100
  20. DefaultHandshakeRetries = 10
  21. DefaultHandshakeTriggerBuffer = 64
  22. DefaultUseRelays = true
  23. )
  24. var (
  25. defaultHandshakeConfig = HandshakeConfig{
  26. tryInterval: DefaultHandshakeTryInterval,
  27. retries: DefaultHandshakeRetries,
  28. triggerBuffer: DefaultHandshakeTriggerBuffer,
  29. useRelays: DefaultUseRelays,
  30. }
  31. )
  32. type HandshakeConfig struct {
  33. tryInterval time.Duration
  34. retries int64
  35. triggerBuffer int
  36. useRelays bool
  37. messageMetrics *MessageMetrics
  38. }
  39. type HandshakeManager struct {
  40. // Mutex for interacting with the vpnIps and indexes maps
  41. sync.RWMutex
  42. vpnIps map[netip.Addr]*HandshakeHostInfo
  43. indexes map[uint32]*HandshakeHostInfo
  44. mainHostMap *HostMap
  45. lightHouse *LightHouse
  46. outside udp.Conn
  47. config HandshakeConfig
  48. OutboundHandshakeTimer *LockingTimerWheel[netip.Addr]
  49. messageMetrics *MessageMetrics
  50. metricInitiated metrics.Counter
  51. metricTimedOut metrics.Counter
  52. f *Interface
  53. l *logrus.Logger
  54. // can be used to trigger outbound handshake for the given vpnIp
  55. trigger chan netip.Addr
  56. }
  57. type HandshakeHostInfo struct {
  58. sync.Mutex
  59. startTime time.Time // Time that we first started trying with this handshake
  60. ready bool // Is the handshake ready
  61. counter int64 // How many attempts have we made so far
  62. lastRemotes []netip.AddrPort // Remotes that we sent to during the previous attempt
  63. packetStore []*cachedPacket // A set of packets to be transmitted once the handshake completes
  64. hostinfo *HostInfo
  65. }
  66. func (hh *HandshakeHostInfo) cachePacket(l *logrus.Logger, t header.MessageType, st header.MessageSubType, packet []byte, f packetCallback, m *cachedPacketMetrics) {
  67. if len(hh.packetStore) < 100 {
  68. tempPacket := make([]byte, len(packet))
  69. copy(tempPacket, packet)
  70. hh.packetStore = append(hh.packetStore, &cachedPacket{t, st, f, tempPacket})
  71. if l.Level >= logrus.DebugLevel {
  72. hh.hostinfo.logger(l).
  73. WithField("length", len(hh.packetStore)).
  74. WithField("stored", true).
  75. Debugf("Packet store")
  76. }
  77. } else {
  78. m.dropped.Inc(1)
  79. if l.Level >= logrus.DebugLevel {
  80. hh.hostinfo.logger(l).
  81. WithField("length", len(hh.packetStore)).
  82. WithField("stored", false).
  83. Debugf("Packet store")
  84. }
  85. }
  86. }
  87. func NewHandshakeManager(l *logrus.Logger, mainHostMap *HostMap, lightHouse *LightHouse, outside udp.Conn, config HandshakeConfig) *HandshakeManager {
  88. return &HandshakeManager{
  89. vpnIps: map[netip.Addr]*HandshakeHostInfo{},
  90. indexes: map[uint32]*HandshakeHostInfo{},
  91. mainHostMap: mainHostMap,
  92. lightHouse: lightHouse,
  93. outside: outside,
  94. config: config,
  95. trigger: make(chan netip.Addr, config.triggerBuffer),
  96. OutboundHandshakeTimer: NewLockingTimerWheel[netip.Addr](config.tryInterval, hsTimeout(config.retries, config.tryInterval)),
  97. messageMetrics: config.messageMetrics,
  98. metricInitiated: metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
  99. metricTimedOut: metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
  100. l: l,
  101. }
  102. }
  103. func (hm *HandshakeManager) Run(ctx context.Context) {
  104. clockSource := time.NewTicker(hm.config.tryInterval)
  105. defer clockSource.Stop()
  106. for {
  107. select {
  108. case <-ctx.Done():
  109. return
  110. case vpnIP := <-hm.trigger:
  111. hm.handleOutbound(vpnIP, true)
  112. case now := <-clockSource.C:
  113. hm.NextOutboundHandshakeTimerTick(now)
  114. }
  115. }
  116. }
  117. func (hm *HandshakeManager) HandleIncoming(addr netip.AddrPort, via *ViaSender, packet []byte, h *header.H) {
  118. // First remote allow list check before we know the vpnIp
  119. if addr.IsValid() {
  120. if !hm.lightHouse.GetRemoteAllowList().AllowUnknownVpnAddr(addr.Addr()) {
  121. hm.l.WithField("udpAddr", addr).Debug("lighthouse.remote_allow_list denied incoming handshake")
  122. return
  123. }
  124. }
  125. switch h.Subtype {
  126. case header.HandshakeIXPSK0:
  127. switch h.MessageCounter {
  128. case 1:
  129. ixHandshakeStage1(hm.f, addr, via, packet, h)
  130. case 2:
  131. newHostinfo := hm.queryIndex(h.RemoteIndex)
  132. tearDown := ixHandshakeStage2(hm.f, addr, via, newHostinfo, packet, h)
  133. if tearDown && newHostinfo != nil {
  134. hm.DeleteHostInfo(newHostinfo.hostinfo)
  135. }
  136. }
  137. }
  138. }
  139. func (hm *HandshakeManager) NextOutboundHandshakeTimerTick(now time.Time) {
  140. hm.OutboundHandshakeTimer.Advance(now)
  141. for {
  142. vpnIp, has := hm.OutboundHandshakeTimer.Purge()
  143. if !has {
  144. break
  145. }
  146. hm.handleOutbound(vpnIp, false)
  147. }
  148. }
  149. func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered bool) {
  150. hh := hm.queryVpnIp(vpnIp)
  151. if hh == nil {
  152. return
  153. }
  154. hh.Lock()
  155. defer hh.Unlock()
  156. hostinfo := hh.hostinfo
  157. // If we are out of time, clean up
  158. if hh.counter >= hm.config.retries {
  159. hh.hostinfo.logger(hm.l).WithField("udpAddrs", hh.hostinfo.remotes.CopyAddrs(hm.mainHostMap.GetPreferredRanges())).
  160. WithField("initiatorIndex", hh.hostinfo.localIndexId).
  161. WithField("remoteIndex", hh.hostinfo.remoteIndexId).
  162. WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
  163. WithField("durationNs", time.Since(hh.startTime).Nanoseconds()).
  164. Info("Handshake timed out")
  165. hm.metricTimedOut.Inc(1)
  166. hm.DeleteHostInfo(hostinfo)
  167. return
  168. }
  169. // Increment the counter to increase our delay, linear backoff
  170. hh.counter++
  171. // Check if we have a handshake packet to transmit yet
  172. if !hh.ready {
  173. if !ixHandshakeStage0(hm.f, hh) {
  174. hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter))
  175. return
  176. }
  177. }
  178. // Get a remotes object if we don't already have one.
  179. // This is mainly to protect us as this should never be the case
  180. // NB ^ This comment doesn't jive. It's how the thing gets initialized.
  181. // It's the common path. Should it update every time, in case a future LH query/queries give us more info?
  182. if hostinfo.remotes == nil {
  183. hostinfo.remotes = hm.lightHouse.QueryCache([]netip.Addr{vpnIp})
  184. }
  185. remotes := hostinfo.remotes.CopyAddrs(hm.mainHostMap.GetPreferredRanges())
  186. remotesHaveChanged := !slices.Equal(remotes, hh.lastRemotes)
  187. // We only care about a lighthouse trigger if we have new remotes to send to.
  188. // This is a very specific optimization for a fast lighthouse reply.
  189. if lighthouseTriggered && !remotesHaveChanged {
  190. // If we didn't return here a lighthouse could cause us to aggressively send handshakes
  191. return
  192. }
  193. hh.lastRemotes = remotes
  194. // This will generate a load of queries for hosts with only 1 ip
  195. // (such as ones registered to the lighthouse with only a private IP)
  196. // So we only do it one time after attempting 5 handshakes already.
  197. if len(remotes) <= 1 && hh.counter == 5 {
  198. // If we only have 1 remote it is highly likely our query raced with the other host registered within the lighthouse
  199. // Our vpnIp here has a tunnel with a lighthouse but has yet to send a host update packet there so we only know about
  200. // the learned public ip for them. Query again to short circuit the promotion counter
  201. hm.lightHouse.QueryServer(vpnIp)
  202. }
  203. // Send the handshake to all known ips, stage 2 takes care of assigning the hostinfo.remote based on the first to reply
  204. var sentTo []netip.AddrPort
  205. hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) {
  206. hm.messageMetrics.Tx(header.Handshake, header.MessageSubType(hostinfo.HandshakePacket[0][1]), 1)
  207. err := hm.outside.WriteTo(hostinfo.HandshakePacket[0], addr)
  208. if err != nil {
  209. hostinfo.logger(hm.l).WithField("udpAddr", addr).
  210. WithField("initiatorIndex", hostinfo.localIndexId).
  211. WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
  212. WithError(err).Error("Failed to send handshake message")
  213. } else {
  214. sentTo = append(sentTo, addr)
  215. }
  216. })
  217. // Don't be too noisy or confusing if we fail to send a handshake - if we don't get through we'll eventually log a timeout,
  218. // so only log when the list of remotes has changed
  219. if remotesHaveChanged {
  220. hostinfo.logger(hm.l).WithField("udpAddrs", sentTo).
  221. WithField("initiatorIndex", hostinfo.localIndexId).
  222. WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
  223. Info("Handshake message sent")
  224. } else if hm.l.Level >= logrus.DebugLevel {
  225. hostinfo.logger(hm.l).WithField("udpAddrs", sentTo).
  226. WithField("initiatorIndex", hostinfo.localIndexId).
  227. WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
  228. Debug("Handshake message sent")
  229. }
  230. if hm.config.useRelays && len(hostinfo.remotes.relays) > 0 {
  231. hostinfo.logger(hm.l).WithField("relays", hostinfo.remotes.relays).Info("Attempt to relay through hosts")
  232. // Send a RelayRequest to all known Relay IP's
  233. for _, relay := range hostinfo.remotes.relays {
  234. // Don't relay to myself
  235. if relay == vpnIp {
  236. continue
  237. }
  238. // Don't relay through the host I'm trying to connect to
  239. if hm.f.myVpnAddrsTable.Contains(relay) {
  240. continue
  241. }
  242. relayHostInfo := hm.mainHostMap.QueryVpnAddr(relay)
  243. if relayHostInfo == nil || !relayHostInfo.remote.IsValid() {
  244. hostinfo.logger(hm.l).WithField("relay", relay.String()).Info("Establish tunnel to relay target")
  245. hm.f.Handshake(relay)
  246. continue
  247. }
  248. // Check the relay HostInfo to see if we already established a relay through
  249. existingRelay, ok := relayHostInfo.relayState.QueryRelayForByIp(vpnIp)
  250. if !ok {
  251. // No relays exist or requested yet.
  252. if relayHostInfo.remote.IsValid() {
  253. idx, err := AddRelay(hm.l, relayHostInfo, hm.mainHostMap, vpnIp, nil, TerminalType, Requested)
  254. if err != nil {
  255. hostinfo.logger(hm.l).WithField("relay", relay.String()).WithError(err).Info("Failed to add relay to hostmap")
  256. continue
  257. }
  258. m := NebulaControl{
  259. Type: NebulaControl_CreateRelayRequest,
  260. InitiatorRelayIndex: idx,
  261. }
  262. switch relayHostInfo.GetCert().Certificate.Version() {
  263. case cert.Version1:
  264. err = buildRelayInfoCertV1(&m, hm.f.myVpnNetworks, vpnIp)
  265. case cert.Version2:
  266. err = buildRelayInfoCertV2(&m, hm.f.myVpnNetworks, vpnIp)
  267. default:
  268. err = errors.New("unknown certificate version found while creating relay")
  269. }
  270. if err != nil {
  271. hostinfo.logger(hm.l).WithError(err).Error("Refusing to relay")
  272. continue
  273. }
  274. msg, err := m.Marshal()
  275. if err != nil {
  276. hostinfo.logger(hm.l).WithError(err).
  277. Error("Failed to marshal Control message to create relay")
  278. } else {
  279. hm.f.SendMessageToHostInfo(header.Control, 0, relayHostInfo, msg, make([]byte, 12), make([]byte, mtu))
  280. hm.l.WithFields(logrus.Fields{
  281. "relayFrom": m.GetRelayFrom(),
  282. "relayTo": vpnIp,
  283. "initiatorRelayIndex": idx,
  284. "relay": relay}).
  285. Info("send CreateRelayRequest")
  286. }
  287. }
  288. continue
  289. }
  290. switch existingRelay.State {
  291. case Established:
  292. hostinfo.logger(hm.l).WithField("relay", relay.String()).Info("Send handshake via relay")
  293. hm.f.SendVia(relayHostInfo, existingRelay, hostinfo.HandshakePacket[0], make([]byte, 12), make([]byte, mtu), false)
  294. case Disestablished:
  295. // Mark this relay as 'requested'
  296. relayHostInfo.relayState.UpdateRelayForByIpState(vpnIp, Requested)
  297. fallthrough
  298. case Requested:
  299. hostinfo.logger(hm.l).WithField("relay", relay.String()).Info("Re-send CreateRelay request")
  300. // Re-send the CreateRelay request, in case the previous one was lost.
  301. m := NebulaControl{
  302. Type: NebulaControl_CreateRelayRequest,
  303. InitiatorRelayIndex: existingRelay.LocalIndex,
  304. }
  305. var err error
  306. switch relayHostInfo.GetCert().Certificate.Version() {
  307. case cert.Version1:
  308. err = buildRelayInfoCertV1(&m, hm.f.myVpnNetworks, vpnIp)
  309. case cert.Version2:
  310. err = buildRelayInfoCertV2(&m, hm.f.myVpnNetworks, vpnIp)
  311. default:
  312. err = errors.New("unknown certificate version found while creating relay")
  313. }
  314. if err != nil {
  315. hostinfo.logger(hm.l).WithError(err).Error("Refusing to relay")
  316. continue
  317. }
  318. msg, err := m.Marshal()
  319. if err != nil {
  320. hostinfo.logger(hm.l).WithError(err).Error("Failed to marshal Control message to create relay")
  321. } else {
  322. // This must send over the hostinfo, not over hm.Hosts[ip]
  323. hm.f.SendMessageToHostInfo(header.Control, 0, relayHostInfo, msg, make([]byte, 12), make([]byte, mtu))
  324. hm.l.WithFields(logrus.Fields{
  325. "relayFrom": m.GetRelayFrom(),
  326. "relayTo": vpnIp,
  327. "initiatorRelayIndex": existingRelay.LocalIndex,
  328. "relay": relay}).
  329. Info("send CreateRelayRequest")
  330. }
  331. case PeerRequested:
  332. // PeerRequested only occurs in Forwarding relays, not Terminal relays, and this is a Terminal relay case.
  333. fallthrough
  334. default:
  335. hostinfo.logger(hm.l).
  336. WithField("vpnIp", vpnIp).
  337. WithField("state", existingRelay.State).
  338. WithField("relay", relay).
  339. Errorf("Relay unexpected state")
  340. }
  341. }
  342. }
  343. // If a lighthouse triggered this attempt then we are still in the timer wheel and do not need to re-add
  344. if !lighthouseTriggered {
  345. hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter))
  346. }
  347. }
  348. // GetOrHandshake will try to find a hostinfo with a fully formed tunnel or start a new handshake if one is not present
  349. // The 2nd argument will be true if the hostinfo is ready to transmit traffic
  350. func (hm *HandshakeManager) GetOrHandshake(vpnIp netip.Addr, cacheCb func(*HandshakeHostInfo)) (*HostInfo, bool) {
  351. hm.mainHostMap.RLock()
  352. h, ok := hm.mainHostMap.Hosts[vpnIp]
  353. hm.mainHostMap.RUnlock()
  354. if ok {
  355. // Do not attempt promotion if you are a lighthouse
  356. if !hm.lightHouse.amLighthouse {
  357. h.TryPromoteBest(hm.mainHostMap.GetPreferredRanges(), hm.f)
  358. }
  359. return h, true
  360. }
  361. return hm.StartHandshake(vpnIp, cacheCb), false
  362. }
  363. // StartHandshake will ensure a handshake is currently being attempted for the provided vpn ip
  364. func (hm *HandshakeManager) StartHandshake(vpnAddr netip.Addr, cacheCb func(*HandshakeHostInfo)) *HostInfo {
  365. hm.Lock()
  366. if hh, ok := hm.vpnIps[vpnAddr]; ok {
  367. // We are already trying to handshake with this vpn ip
  368. if cacheCb != nil {
  369. cacheCb(hh)
  370. }
  371. hm.Unlock()
  372. return hh.hostinfo
  373. }
  374. hostinfo := &HostInfo{
  375. vpnAddrs: []netip.Addr{vpnAddr},
  376. HandshakePacket: make(map[uint8][]byte, 0),
  377. relayState: RelayState{
  378. relays: nil,
  379. relayForByAddr: map[netip.Addr]*Relay{},
  380. relayForByIdx: map[uint32]*Relay{},
  381. },
  382. }
  383. hh := &HandshakeHostInfo{
  384. hostinfo: hostinfo,
  385. startTime: time.Now(),
  386. }
  387. hm.vpnIps[vpnAddr] = hh
  388. hm.metricInitiated.Inc(1)
  389. hm.OutboundHandshakeTimer.Add(vpnAddr, hm.config.tryInterval)
  390. if cacheCb != nil {
  391. cacheCb(hh)
  392. }
  393. // If this is a static host, we don't need to wait for the HostQueryReply
  394. // We can trigger the handshake right now
  395. _, doTrigger := hm.lightHouse.GetStaticHostList()[vpnAddr]
  396. if !doTrigger {
  397. // Add any calculated remotes, and trigger early handshake if one found
  398. doTrigger = hm.lightHouse.addCalculatedRemotes(vpnAddr)
  399. }
  400. if doTrigger {
  401. select {
  402. case hm.trigger <- vpnAddr:
  403. default:
  404. }
  405. }
  406. hm.Unlock()
  407. hm.lightHouse.QueryServer(vpnAddr)
  408. return hostinfo
  409. }
  410. var (
  411. ErrExistingHostInfo = errors.New("existing hostinfo")
  412. ErrAlreadySeen = errors.New("already seen")
  413. ErrLocalIndexCollision = errors.New("local index collision")
  414. )
  415. // CheckAndComplete checks for any conflicts in the main and pending hostmap
  416. // before adding hostinfo to main. If err is nil, it was added. Otherwise err will be:
  417. //
  418. // ErrAlreadySeen if we already have an entry in the hostmap that has seen the
  419. // exact same handshake packet
  420. //
  421. // ErrExistingHostInfo if we already have an entry in the hostmap for this
  422. // VpnIp and the new handshake was older than the one we currently have
  423. //
  424. // ErrLocalIndexCollision if we already have an entry in the main or pending
  425. // hostmap for the hostinfo.localIndexId.
  426. func (hm *HandshakeManager) CheckAndComplete(hostinfo *HostInfo, handshakePacket uint8, f *Interface) (*HostInfo, error) {
  427. hm.mainHostMap.Lock()
  428. defer hm.mainHostMap.Unlock()
  429. hm.Lock()
  430. defer hm.Unlock()
  431. // Check if we already have a tunnel with this vpn ip
  432. existingHostInfo, found := hm.mainHostMap.Hosts[hostinfo.vpnAddrs[0]]
  433. if found && existingHostInfo != nil {
  434. testHostInfo := existingHostInfo
  435. for testHostInfo != nil {
  436. // Is it just a delayed handshake packet?
  437. if bytes.Equal(hostinfo.HandshakePacket[handshakePacket], testHostInfo.HandshakePacket[handshakePacket]) {
  438. return testHostInfo, ErrAlreadySeen
  439. }
  440. testHostInfo = testHostInfo.next
  441. }
  442. // Is this a newer handshake?
  443. if existingHostInfo.lastHandshakeTime >= hostinfo.lastHandshakeTime && !existingHostInfo.ConnectionState.initiator {
  444. return existingHostInfo, ErrExistingHostInfo
  445. }
  446. existingHostInfo.logger(hm.l).Info("Taking new handshake")
  447. }
  448. existingIndex, found := hm.mainHostMap.Indexes[hostinfo.localIndexId]
  449. if found {
  450. // We have a collision, but for a different hostinfo
  451. return existingIndex, ErrLocalIndexCollision
  452. }
  453. existingPendingIndex, found := hm.indexes[hostinfo.localIndexId]
  454. if found && existingPendingIndex.hostinfo != hostinfo {
  455. // We have a collision, but for a different hostinfo
  456. return existingPendingIndex.hostinfo, ErrLocalIndexCollision
  457. }
  458. existingRemoteIndex, found := hm.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId]
  459. if found && existingRemoteIndex != nil && existingRemoteIndex.vpnAddrs[0] != hostinfo.vpnAddrs[0] {
  460. // We have a collision, but this can happen since we can't control
  461. // the remote ID. Just log about the situation as a note.
  462. hostinfo.logger(hm.l).
  463. WithField("remoteIndex", hostinfo.remoteIndexId).WithField("collision", existingRemoteIndex.vpnAddrs).
  464. Info("New host shadows existing host remoteIndex")
  465. }
  466. hm.mainHostMap.unlockedAddHostInfo(hostinfo, f)
  467. return existingHostInfo, nil
  468. }
  469. // Complete is a simpler version of CheckAndComplete when we already know we
  470. // won't have a localIndexId collision because we already have an entry in the
  471. // pendingHostMap. An existing hostinfo is returned if there was one.
  472. func (hm *HandshakeManager) Complete(hostinfo *HostInfo, f *Interface) {
  473. hm.mainHostMap.Lock()
  474. defer hm.mainHostMap.Unlock()
  475. hm.Lock()
  476. defer hm.Unlock()
  477. existingRemoteIndex, found := hm.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId]
  478. if found && existingRemoteIndex != nil {
  479. // We have a collision, but this can happen since we can't control
  480. // the remote ID. Just log about the situation as a note.
  481. hostinfo.logger(hm.l).
  482. WithField("remoteIndex", hostinfo.remoteIndexId).WithField("collision", existingRemoteIndex.vpnAddrs).
  483. Info("New host shadows existing host remoteIndex")
  484. }
  485. // We need to remove from the pending hostmap first to avoid undoing work when after to the main hostmap.
  486. hm.unlockedDeleteHostInfo(hostinfo)
  487. hm.mainHostMap.unlockedAddHostInfo(hostinfo, f)
  488. }
  489. // allocateIndex generates a unique localIndexId for this HostInfo
  490. // and adds it to the pendingHostMap. Will error if we are unable to generate
  491. // a unique localIndexId
  492. func (hm *HandshakeManager) allocateIndex(hh *HandshakeHostInfo) error {
  493. hm.mainHostMap.RLock()
  494. defer hm.mainHostMap.RUnlock()
  495. hm.Lock()
  496. defer hm.Unlock()
  497. for i := 0; i < 32; i++ {
  498. index, err := generateIndex(hm.l)
  499. if err != nil {
  500. return err
  501. }
  502. _, inPending := hm.indexes[index]
  503. _, inMain := hm.mainHostMap.Indexes[index]
  504. if !inMain && !inPending {
  505. hh.hostinfo.localIndexId = index
  506. hm.indexes[index] = hh
  507. return nil
  508. }
  509. }
  510. return errors.New("failed to generate unique localIndexId")
  511. }
  512. func (hm *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) {
  513. hm.Lock()
  514. defer hm.Unlock()
  515. hm.unlockedDeleteHostInfo(hostinfo)
  516. }
  517. func (hm *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) {
  518. for _, addr := range hostinfo.vpnAddrs {
  519. delete(hm.vpnIps, addr)
  520. }
  521. if len(hm.vpnIps) == 0 {
  522. hm.vpnIps = map[netip.Addr]*HandshakeHostInfo{}
  523. }
  524. delete(hm.indexes, hostinfo.localIndexId)
  525. if len(hm.indexes) == 0 {
  526. hm.indexes = map[uint32]*HandshakeHostInfo{}
  527. }
  528. if hm.l.Level >= logrus.DebugLevel {
  529. hm.l.WithField("hostMap", m{"mapTotalSize": len(hm.vpnIps),
  530. "vpnAddrs": hostinfo.vpnAddrs, "indexNumber": hostinfo.localIndexId, "remoteIndexNumber": hostinfo.remoteIndexId}).
  531. Debug("Pending hostmap hostInfo deleted")
  532. }
  533. }
  534. func (hm *HandshakeManager) QueryVpnAddr(vpnIp netip.Addr) *HostInfo {
  535. hh := hm.queryVpnIp(vpnIp)
  536. if hh != nil {
  537. return hh.hostinfo
  538. }
  539. return nil
  540. }
  541. func (hm *HandshakeManager) queryVpnIp(vpnIp netip.Addr) *HandshakeHostInfo {
  542. hm.RLock()
  543. defer hm.RUnlock()
  544. return hm.vpnIps[vpnIp]
  545. }
  546. func (hm *HandshakeManager) QueryIndex(index uint32) *HostInfo {
  547. hh := hm.queryIndex(index)
  548. if hh != nil {
  549. return hh.hostinfo
  550. }
  551. return nil
  552. }
  553. func (hm *HandshakeManager) queryIndex(index uint32) *HandshakeHostInfo {
  554. hm.RLock()
  555. defer hm.RUnlock()
  556. return hm.indexes[index]
  557. }
  558. func (hm *HandshakeManager) GetPreferredRanges() []netip.Prefix {
  559. return hm.mainHostMap.GetPreferredRanges()
  560. }
  561. func (hm *HandshakeManager) ForEachVpnAddr(f controlEach) {
  562. hm.RLock()
  563. defer hm.RUnlock()
  564. for _, v := range hm.vpnIps {
  565. f(v.hostinfo)
  566. }
  567. }
  568. func (hm *HandshakeManager) ForEachIndex(f controlEach) {
  569. hm.RLock()
  570. defer hm.RUnlock()
  571. for _, v := range hm.indexes {
  572. f(v.hostinfo)
  573. }
  574. }
  575. func (hm *HandshakeManager) EmitStats() {
  576. hm.RLock()
  577. hostLen := len(hm.vpnIps)
  578. indexLen := len(hm.indexes)
  579. hm.RUnlock()
  580. metrics.GetOrRegisterGauge("hostmap.pending.hosts", nil).Update(int64(hostLen))
  581. metrics.GetOrRegisterGauge("hostmap.pending.indexes", nil).Update(int64(indexLen))
  582. hm.mainHostMap.EmitStats()
  583. }
  584. // Utility functions below
  585. func generateIndex(l *logrus.Logger) (uint32, error) {
  586. b := make([]byte, 4)
  587. // Let zero mean we don't know the ID, so don't generate zero
  588. var index uint32
  589. for index == 0 {
  590. _, err := rand.Read(b)
  591. if err != nil {
  592. l.Errorln(err)
  593. return 0, err
  594. }
  595. index = binary.BigEndian.Uint32(b)
  596. }
  597. if l.Level >= logrus.DebugLevel {
  598. l.WithField("index", index).
  599. Debug("Generated index")
  600. }
  601. return index, nil
  602. }
  603. func hsTimeout(tries int64, interval time.Duration) time.Duration {
  604. return time.Duration(tries / 2 * ((2 * int64(interval)) + (tries-1)*int64(interval)))
  605. }
  606. var errNoRelayTooOld = errors.New("can not establish v1 relay with a v6 network because the relay is not running a current nebula version")
  607. func buildRelayInfoCertV1(m *NebulaControl, myVpnNetworks []netip.Prefix, peerVpnIp netip.Addr) error {
  608. relayFrom := myVpnNetworks[0].Addr()
  609. if !relayFrom.Is4() {
  610. return errNoRelayTooOld
  611. }
  612. if !peerVpnIp.Is4() {
  613. return errNoRelayTooOld
  614. }
  615. b := relayFrom.As4()
  616. m.OldRelayFromAddr = binary.BigEndian.Uint32(b[:])
  617. b = peerVpnIp.As4()
  618. m.OldRelayToAddr = binary.BigEndian.Uint32(b[:])
  619. return nil
  620. }
  621. func buildRelayInfoCertV2(m *NebulaControl, myVpnNetworks []netip.Prefix, peerVpnIp netip.Addr) error {
  622. for i := range myVpnNetworks {
  623. if myVpnNetworks[i].Contains(peerVpnIp) {
  624. m.RelayFromAddr = netAddrToProtoAddr(myVpnNetworks[i].Addr())
  625. m.RelayToAddr = netAddrToProtoAddr(peerVpnIp)
  626. return nil
  627. }
  628. }
  629. return errors.New("cannot establish relay, no networks in common")
  630. }