handshake_manager.go 21 KB

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