handshake_manager.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670
  1. package nebula
  2. import (
  3. "bytes"
  4. "context"
  5. "crypto/rand"
  6. "encoding/binary"
  7. "errors"
  8. "net/netip"
  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/udp"
  15. "golang.org/x/exp/slices"
  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 int64
  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[netip.Addr]*HandshakeHostInfo
  42. indexes map[uint32]*HandshakeHostInfo
  43. mainHostMap *HostMap
  44. lightHouse *LightHouse
  45. outside udp.Conn
  46. config HandshakeConfig
  47. OutboundHandshakeTimer *LockingTimerWheel[netip.Addr]
  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 netip.Addr
  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 int64 // How many attempts have we made so far
  61. lastRemotes []netip.AddrPort // 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[netip.Addr]*HandshakeHostInfo{},
  89. indexes: map[uint32]*HandshakeHostInfo{},
  90. mainHostMap: mainHostMap,
  91. lightHouse: lightHouse,
  92. outside: outside,
  93. config: config,
  94. trigger: make(chan netip.Addr, config.triggerBuffer),
  95. OutboundHandshakeTimer: NewLockingTimerWheel[netip.Addr](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 netip.AddrPort, via *ViaSender, packet []byte, h *header.H) {
  117. // First remote allow list check before we know the vpnIp
  118. if addr.IsValid() {
  119. if !hm.lightHouse.GetRemoteAllowList().AllowUnknownVpnIp(addr.Addr()) {
  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 netip.Addr, 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.GetPreferredRanges())).
  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.GetPreferredRanges())
  185. remotesHaveChanged := !slices.Equal(remotes, 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 []netip.AddrPort
  204. hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ 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.myVpnNet.Addr() {
  235. continue
  236. }
  237. relayHostInfo := hm.mainHostMap.QueryVpnIp(relay)
  238. if relayHostInfo == nil || !relayHostInfo.remote.IsValid() {
  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. //TODO: IPV6-WORK
  252. myVpnIpB := hm.f.myVpnNet.Addr().As4()
  253. theirVpnIpB := vpnIp.As4()
  254. // Re-send the CreateRelay request, in case the previous one was lost.
  255. m := NebulaControl{
  256. Type: NebulaControl_CreateRelayRequest,
  257. InitiatorRelayIndex: existingRelay.LocalIndex,
  258. RelayFromIp: binary.BigEndian.Uint32(myVpnIpB[:]),
  259. RelayToIp: binary.BigEndian.Uint32(theirVpnIpB[:]),
  260. }
  261. msg, err := m.Marshal()
  262. if err != nil {
  263. hostinfo.logger(hm.l).
  264. WithError(err).
  265. Error("Failed to marshal Control message to create relay")
  266. } else {
  267. // This must send over the hostinfo, not over hm.Hosts[ip]
  268. hm.f.SendMessageToHostInfo(header.Control, 0, relayHostInfo, msg, make([]byte, 12), make([]byte, mtu))
  269. hm.l.WithFields(logrus.Fields{
  270. "relayFrom": hm.f.myVpnNet.Addr(),
  271. "relayTo": vpnIp,
  272. "initiatorRelayIndex": existingRelay.LocalIndex,
  273. "relay": relay}).
  274. Info("send CreateRelayRequest")
  275. }
  276. default:
  277. hostinfo.logger(hm.l).
  278. WithField("vpnIp", vpnIp).
  279. WithField("state", existingRelay.State).
  280. WithField("relay", relayHostInfo.vpnIp).
  281. Errorf("Relay unexpected state")
  282. }
  283. } else {
  284. // No relays exist or requested yet.
  285. if relayHostInfo.remote.IsValid() {
  286. idx, err := AddRelay(hm.l, relayHostInfo, hm.mainHostMap, vpnIp, nil, TerminalType, Requested)
  287. if err != nil {
  288. hostinfo.logger(hm.l).WithField("relay", relay.String()).WithError(err).Info("Failed to add relay to hostmap")
  289. }
  290. //TODO: IPV6-WORK
  291. myVpnIpB := hm.f.myVpnNet.Addr().As4()
  292. theirVpnIpB := vpnIp.As4()
  293. m := NebulaControl{
  294. Type: NebulaControl_CreateRelayRequest,
  295. InitiatorRelayIndex: idx,
  296. RelayFromIp: binary.BigEndian.Uint32(myVpnIpB[:]),
  297. RelayToIp: binary.BigEndian.Uint32(theirVpnIpB[:]),
  298. }
  299. msg, err := m.Marshal()
  300. if err != nil {
  301. hostinfo.logger(hm.l).
  302. WithError(err).
  303. Error("Failed to marshal Control message to create relay")
  304. } else {
  305. hm.f.SendMessageToHostInfo(header.Control, 0, relayHostInfo, msg, make([]byte, 12), make([]byte, mtu))
  306. hm.l.WithFields(logrus.Fields{
  307. "relayFrom": hm.f.myVpnNet.Addr(),
  308. "relayTo": vpnIp,
  309. "initiatorRelayIndex": idx,
  310. "relay": relay}).
  311. Info("send CreateRelayRequest")
  312. }
  313. }
  314. }
  315. }
  316. }
  317. // If a lighthouse triggered this attempt then we are still in the timer wheel and do not need to re-add
  318. if !lighthouseTriggered {
  319. hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval*time.Duration(hh.counter))
  320. }
  321. }
  322. // GetOrHandshake will try to find a hostinfo with a fully formed tunnel or start a new handshake if one is not present
  323. // The 2nd argument will be true if the hostinfo is ready to transmit traffic
  324. func (hm *HandshakeManager) GetOrHandshake(vpnIp netip.Addr, cacheCb func(*HandshakeHostInfo)) (*HostInfo, bool) {
  325. hm.mainHostMap.RLock()
  326. h, ok := hm.mainHostMap.Hosts[vpnIp]
  327. hm.mainHostMap.RUnlock()
  328. if ok {
  329. // Do not attempt promotion if you are a lighthouse
  330. if !hm.lightHouse.amLighthouse {
  331. h.TryPromoteBest(hm.mainHostMap.GetPreferredRanges(), hm.f)
  332. }
  333. return h, true
  334. }
  335. return hm.StartHandshake(vpnIp, cacheCb), false
  336. }
  337. // StartHandshake will ensure a handshake is currently being attempted for the provided vpn ip
  338. func (hm *HandshakeManager) StartHandshake(vpnIp netip.Addr, cacheCb func(*HandshakeHostInfo)) *HostInfo {
  339. hm.Lock()
  340. if hh, ok := hm.vpnIps[vpnIp]; ok {
  341. // We are already trying to handshake with this vpn ip
  342. if cacheCb != nil {
  343. cacheCb(hh)
  344. }
  345. hm.Unlock()
  346. return hh.hostinfo
  347. }
  348. hostinfo := &HostInfo{
  349. vpnIp: vpnIp,
  350. HandshakePacket: make(map[uint8][]byte, 0),
  351. relayState: RelayState{
  352. relays: map[netip.Addr]struct{}{},
  353. relayForByIp: map[netip.Addr]*Relay{},
  354. relayForByIdx: map[uint32]*Relay{},
  355. },
  356. }
  357. hh := &HandshakeHostInfo{
  358. hostinfo: hostinfo,
  359. startTime: time.Now(),
  360. }
  361. hm.vpnIps[vpnIp] = hh
  362. hm.metricInitiated.Inc(1)
  363. hm.OutboundHandshakeTimer.Add(vpnIp, hm.config.tryInterval)
  364. if cacheCb != nil {
  365. cacheCb(hh)
  366. }
  367. // If this is a static host, we don't need to wait for the HostQueryReply
  368. // We can trigger the handshake right now
  369. _, doTrigger := hm.lightHouse.GetStaticHostList()[vpnIp]
  370. if !doTrigger {
  371. // Add any calculated remotes, and trigger early handshake if one found
  372. doTrigger = hm.lightHouse.addCalculatedRemotes(vpnIp)
  373. }
  374. if doTrigger {
  375. select {
  376. case hm.trigger <- vpnIp:
  377. default:
  378. }
  379. }
  380. hm.Unlock()
  381. hm.lightHouse.QueryServer(vpnIp)
  382. return hostinfo
  383. }
  384. var (
  385. ErrExistingHostInfo = errors.New("existing hostinfo")
  386. ErrAlreadySeen = errors.New("already seen")
  387. ErrLocalIndexCollision = errors.New("local index collision")
  388. )
  389. // CheckAndComplete checks for any conflicts in the main and pending hostmap
  390. // before adding hostinfo to main. If err is nil, it was added. Otherwise err will be:
  391. //
  392. // ErrAlreadySeen if we already have an entry in the hostmap that has seen the
  393. // exact same handshake packet
  394. //
  395. // ErrExistingHostInfo if we already have an entry in the hostmap for this
  396. // VpnIp and the new handshake was older than the one we currently have
  397. //
  398. // ErrLocalIndexCollision if we already have an entry in the main or pending
  399. // hostmap for the hostinfo.localIndexId.
  400. func (c *HandshakeManager) CheckAndComplete(hostinfo *HostInfo, handshakePacket uint8, f *Interface) (*HostInfo, error) {
  401. c.mainHostMap.Lock()
  402. defer c.mainHostMap.Unlock()
  403. c.Lock()
  404. defer c.Unlock()
  405. // Check if we already have a tunnel with this vpn ip
  406. existingHostInfo, found := c.mainHostMap.Hosts[hostinfo.vpnIp]
  407. if found && existingHostInfo != nil {
  408. testHostInfo := existingHostInfo
  409. for testHostInfo != nil {
  410. // Is it just a delayed handshake packet?
  411. if bytes.Equal(hostinfo.HandshakePacket[handshakePacket], testHostInfo.HandshakePacket[handshakePacket]) {
  412. return testHostInfo, ErrAlreadySeen
  413. }
  414. testHostInfo = testHostInfo.next
  415. }
  416. // Is this a newer handshake?
  417. if existingHostInfo.lastHandshakeTime >= hostinfo.lastHandshakeTime && !existingHostInfo.ConnectionState.initiator {
  418. return existingHostInfo, ErrExistingHostInfo
  419. }
  420. existingHostInfo.logger(c.l).Info("Taking new handshake")
  421. }
  422. existingIndex, found := c.mainHostMap.Indexes[hostinfo.localIndexId]
  423. if found {
  424. // We have a collision, but for a different hostinfo
  425. return existingIndex, ErrLocalIndexCollision
  426. }
  427. existingPendingIndex, found := c.indexes[hostinfo.localIndexId]
  428. if found && existingPendingIndex.hostinfo != hostinfo {
  429. // We have a collision, but for a different hostinfo
  430. return existingPendingIndex.hostinfo, ErrLocalIndexCollision
  431. }
  432. existingRemoteIndex, found := c.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId]
  433. if found && existingRemoteIndex != nil && existingRemoteIndex.vpnIp != hostinfo.vpnIp {
  434. // We have a collision, but this can happen since we can't control
  435. // the remote ID. Just log about the situation as a note.
  436. hostinfo.logger(c.l).
  437. WithField("remoteIndex", hostinfo.remoteIndexId).WithField("collision", existingRemoteIndex.vpnIp).
  438. Info("New host shadows existing host remoteIndex")
  439. }
  440. c.mainHostMap.unlockedAddHostInfo(hostinfo, f)
  441. return existingHostInfo, nil
  442. }
  443. // Complete is a simpler version of CheckAndComplete when we already know we
  444. // won't have a localIndexId collision because we already have an entry in the
  445. // pendingHostMap. An existing hostinfo is returned if there was one.
  446. func (hm *HandshakeManager) Complete(hostinfo *HostInfo, f *Interface) {
  447. hm.mainHostMap.Lock()
  448. defer hm.mainHostMap.Unlock()
  449. hm.Lock()
  450. defer hm.Unlock()
  451. existingRemoteIndex, found := hm.mainHostMap.RemoteIndexes[hostinfo.remoteIndexId]
  452. if found && existingRemoteIndex != nil {
  453. // We have a collision, but this can happen since we can't control
  454. // the remote ID. Just log about the situation as a note.
  455. hostinfo.logger(hm.l).
  456. WithField("remoteIndex", hostinfo.remoteIndexId).WithField("collision", existingRemoteIndex.vpnIp).
  457. Info("New host shadows existing host remoteIndex")
  458. }
  459. // We need to remove from the pending hostmap first to avoid undoing work when after to the main hostmap.
  460. hm.unlockedDeleteHostInfo(hostinfo)
  461. hm.mainHostMap.unlockedAddHostInfo(hostinfo, f)
  462. }
  463. // allocateIndex generates a unique localIndexId for this HostInfo
  464. // and adds it to the pendingHostMap. Will error if we are unable to generate
  465. // a unique localIndexId
  466. func (hm *HandshakeManager) allocateIndex(hh *HandshakeHostInfo) error {
  467. hm.mainHostMap.RLock()
  468. defer hm.mainHostMap.RUnlock()
  469. hm.Lock()
  470. defer hm.Unlock()
  471. for i := 0; i < 32; i++ {
  472. index, err := generateIndex(hm.l)
  473. if err != nil {
  474. return err
  475. }
  476. _, inPending := hm.indexes[index]
  477. _, inMain := hm.mainHostMap.Indexes[index]
  478. if !inMain && !inPending {
  479. hh.hostinfo.localIndexId = index
  480. hm.indexes[index] = hh
  481. return nil
  482. }
  483. }
  484. return errors.New("failed to generate unique localIndexId")
  485. }
  486. func (c *HandshakeManager) DeleteHostInfo(hostinfo *HostInfo) {
  487. c.Lock()
  488. defer c.Unlock()
  489. c.unlockedDeleteHostInfo(hostinfo)
  490. }
  491. func (c *HandshakeManager) unlockedDeleteHostInfo(hostinfo *HostInfo) {
  492. delete(c.vpnIps, hostinfo.vpnIp)
  493. if len(c.vpnIps) == 0 {
  494. c.vpnIps = map[netip.Addr]*HandshakeHostInfo{}
  495. }
  496. delete(c.indexes, hostinfo.localIndexId)
  497. if len(c.vpnIps) == 0 {
  498. c.indexes = map[uint32]*HandshakeHostInfo{}
  499. }
  500. if c.l.Level >= logrus.DebugLevel {
  501. c.l.WithField("hostMap", m{"mapTotalSize": len(c.vpnIps),
  502. "vpnIp": hostinfo.vpnIp, "indexNumber": hostinfo.localIndexId, "remoteIndexNumber": hostinfo.remoteIndexId}).
  503. Debug("Pending hostmap hostInfo deleted")
  504. }
  505. }
  506. func (hm *HandshakeManager) QueryVpnIp(vpnIp netip.Addr) *HostInfo {
  507. hh := hm.queryVpnIp(vpnIp)
  508. if hh != nil {
  509. return hh.hostinfo
  510. }
  511. return nil
  512. }
  513. func (hm *HandshakeManager) queryVpnIp(vpnIp netip.Addr) *HandshakeHostInfo {
  514. hm.RLock()
  515. defer hm.RUnlock()
  516. return hm.vpnIps[vpnIp]
  517. }
  518. func (hm *HandshakeManager) QueryIndex(index uint32) *HostInfo {
  519. hh := hm.queryIndex(index)
  520. if hh != nil {
  521. return hh.hostinfo
  522. }
  523. return nil
  524. }
  525. func (hm *HandshakeManager) queryIndex(index uint32) *HandshakeHostInfo {
  526. hm.RLock()
  527. defer hm.RUnlock()
  528. return hm.indexes[index]
  529. }
  530. func (c *HandshakeManager) GetPreferredRanges() []netip.Prefix {
  531. return c.mainHostMap.GetPreferredRanges()
  532. }
  533. func (c *HandshakeManager) ForEachVpnIp(f controlEach) {
  534. c.RLock()
  535. defer c.RUnlock()
  536. for _, v := range c.vpnIps {
  537. f(v.hostinfo)
  538. }
  539. }
  540. func (c *HandshakeManager) ForEachIndex(f controlEach) {
  541. c.RLock()
  542. defer c.RUnlock()
  543. for _, v := range c.indexes {
  544. f(v.hostinfo)
  545. }
  546. }
  547. func (c *HandshakeManager) EmitStats() {
  548. c.RLock()
  549. hostLen := len(c.vpnIps)
  550. indexLen := len(c.indexes)
  551. c.RUnlock()
  552. metrics.GetOrRegisterGauge("hostmap.pending.hosts", nil).Update(int64(hostLen))
  553. metrics.GetOrRegisterGauge("hostmap.pending.indexes", nil).Update(int64(indexLen))
  554. c.mainHostMap.EmitStats()
  555. }
  556. // Utility functions below
  557. func generateIndex(l *logrus.Logger) (uint32, error) {
  558. b := make([]byte, 4)
  559. // Let zero mean we don't know the ID, so don't generate zero
  560. var index uint32
  561. for index == 0 {
  562. _, err := rand.Read(b)
  563. if err != nil {
  564. l.Errorln(err)
  565. return 0, err
  566. }
  567. index = binary.BigEndian.Uint32(b)
  568. }
  569. if l.Level >= logrus.DebugLevel {
  570. l.WithField("index", index).
  571. Debug("Generated index")
  572. }
  573. return index, nil
  574. }
  575. func hsTimeout(tries int64, interval time.Duration) time.Duration {
  576. return time.Duration(tries / 2 * ((2 * int64(interval)) + (tries-1)*int64(interval)))
  577. }