connection_manager.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485
  1. package nebula
  2. import (
  3. "bytes"
  4. "context"
  5. "sync"
  6. "time"
  7. "github.com/rcrowley/go-metrics"
  8. "github.com/sirupsen/logrus"
  9. "github.com/slackhq/nebula/cert"
  10. "github.com/slackhq/nebula/header"
  11. "github.com/slackhq/nebula/iputil"
  12. "github.com/slackhq/nebula/udp"
  13. )
  14. type trafficDecision int
  15. const (
  16. doNothing trafficDecision = 0
  17. deleteTunnel trafficDecision = 1 // delete the hostinfo on our side, do not notify the remote
  18. closeTunnel trafficDecision = 2 // delete the hostinfo and notify the remote
  19. swapPrimary trafficDecision = 3
  20. migrateRelays trafficDecision = 4
  21. )
  22. type connectionManager struct {
  23. in map[uint32]struct{}
  24. inLock *sync.RWMutex
  25. out map[uint32]struct{}
  26. outLock *sync.RWMutex
  27. // relayUsed holds which relay localIndexs are in use
  28. relayUsed map[uint32]struct{}
  29. relayUsedLock *sync.RWMutex
  30. hostMap *HostMap
  31. trafficTimer *LockingTimerWheel[uint32]
  32. intf *Interface
  33. pendingDeletion map[uint32]struct{}
  34. punchy *Punchy
  35. checkInterval time.Duration
  36. pendingDeletionInterval time.Duration
  37. metricsTxPunchy metrics.Counter
  38. l *logrus.Logger
  39. }
  40. func newConnectionManager(ctx context.Context, l *logrus.Logger, intf *Interface, checkInterval, pendingDeletionInterval time.Duration, punchy *Punchy) *connectionManager {
  41. var max time.Duration
  42. if checkInterval < pendingDeletionInterval {
  43. max = pendingDeletionInterval
  44. } else {
  45. max = checkInterval
  46. }
  47. nc := &connectionManager{
  48. hostMap: intf.hostMap,
  49. in: make(map[uint32]struct{}),
  50. inLock: &sync.RWMutex{},
  51. out: make(map[uint32]struct{}),
  52. outLock: &sync.RWMutex{},
  53. relayUsed: make(map[uint32]struct{}),
  54. relayUsedLock: &sync.RWMutex{},
  55. trafficTimer: NewLockingTimerWheel[uint32](time.Millisecond*500, max),
  56. intf: intf,
  57. pendingDeletion: make(map[uint32]struct{}),
  58. checkInterval: checkInterval,
  59. pendingDeletionInterval: pendingDeletionInterval,
  60. punchy: punchy,
  61. metricsTxPunchy: metrics.GetOrRegisterCounter("messages.tx.punchy", nil),
  62. l: l,
  63. }
  64. nc.Start(ctx)
  65. return nc
  66. }
  67. func (n *connectionManager) In(localIndex uint32) {
  68. n.inLock.RLock()
  69. // If this already exists, return
  70. if _, ok := n.in[localIndex]; ok {
  71. n.inLock.RUnlock()
  72. return
  73. }
  74. n.inLock.RUnlock()
  75. n.inLock.Lock()
  76. n.in[localIndex] = struct{}{}
  77. n.inLock.Unlock()
  78. }
  79. func (n *connectionManager) Out(localIndex uint32) {
  80. n.outLock.RLock()
  81. // If this already exists, return
  82. if _, ok := n.out[localIndex]; ok {
  83. n.outLock.RUnlock()
  84. return
  85. }
  86. n.outLock.RUnlock()
  87. n.outLock.Lock()
  88. n.out[localIndex] = struct{}{}
  89. n.outLock.Unlock()
  90. }
  91. func (n *connectionManager) RelayUsed(localIndex uint32) {
  92. n.relayUsedLock.RLock()
  93. // If this already exists, return
  94. if _, ok := n.relayUsed[localIndex]; ok {
  95. n.relayUsedLock.RUnlock()
  96. return
  97. }
  98. n.relayUsedLock.RUnlock()
  99. n.relayUsedLock.Lock()
  100. n.relayUsed[localIndex] = struct{}{}
  101. n.relayUsedLock.Unlock()
  102. }
  103. // getAndResetTrafficCheck returns if there was any inbound or outbound traffic within the last tick and
  104. // resets the state for this local index
  105. func (n *connectionManager) getAndResetTrafficCheck(localIndex uint32) (bool, bool) {
  106. n.inLock.Lock()
  107. n.outLock.Lock()
  108. _, in := n.in[localIndex]
  109. _, out := n.out[localIndex]
  110. delete(n.in, localIndex)
  111. delete(n.out, localIndex)
  112. n.inLock.Unlock()
  113. n.outLock.Unlock()
  114. return in, out
  115. }
  116. func (n *connectionManager) AddTrafficWatch(localIndex uint32) {
  117. // Use a write lock directly because it should be incredibly rare that we are ever already tracking this index
  118. n.outLock.Lock()
  119. if _, ok := n.out[localIndex]; ok {
  120. n.outLock.Unlock()
  121. return
  122. }
  123. n.out[localIndex] = struct{}{}
  124. n.trafficTimer.Add(localIndex, n.checkInterval)
  125. n.outLock.Unlock()
  126. }
  127. func (n *connectionManager) Start(ctx context.Context) {
  128. go n.Run(ctx)
  129. }
  130. func (n *connectionManager) Run(ctx context.Context) {
  131. //TODO: this tick should be based on the min wheel tick? Check firewall
  132. clockSource := time.NewTicker(500 * time.Millisecond)
  133. defer clockSource.Stop()
  134. p := []byte("")
  135. nb := make([]byte, 12, 12)
  136. out := make([]byte, mtu)
  137. for {
  138. select {
  139. case <-ctx.Done():
  140. return
  141. case now := <-clockSource.C:
  142. n.trafficTimer.Advance(now)
  143. for {
  144. localIndex, has := n.trafficTimer.Purge()
  145. if !has {
  146. break
  147. }
  148. n.doTrafficCheck(localIndex, p, nb, out, now)
  149. }
  150. }
  151. }
  152. }
  153. func (n *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte, now time.Time) {
  154. decision, hostinfo, primary := n.makeTrafficDecision(localIndex, p, nb, out, now)
  155. switch decision {
  156. case deleteTunnel:
  157. if n.hostMap.DeleteHostInfo(hostinfo) {
  158. // Only clearing the lighthouse cache if this is the last hostinfo for this vpn ip in the hostmap
  159. n.intf.lightHouse.DeleteVpnIp(hostinfo.vpnIp)
  160. }
  161. case closeTunnel:
  162. n.intf.sendCloseTunnel(hostinfo)
  163. n.intf.closeTunnel(hostinfo)
  164. case swapPrimary:
  165. n.swapPrimary(hostinfo, primary)
  166. case migrateRelays:
  167. n.migrateRelayUsed(hostinfo, primary)
  168. }
  169. n.resetRelayTrafficCheck(hostinfo)
  170. }
  171. func (n *connectionManager) resetRelayTrafficCheck(hostinfo *HostInfo) {
  172. if hostinfo != nil {
  173. n.relayUsedLock.Lock()
  174. defer n.relayUsedLock.Unlock()
  175. // No need to migrate any relays, delete usage info now.
  176. for _, idx := range hostinfo.relayState.CopyRelayForIdxs() {
  177. delete(n.relayUsed, idx)
  178. }
  179. }
  180. }
  181. func (n *connectionManager) migrateRelayUsed(oldhostinfo, newhostinfo *HostInfo) {
  182. relayFor := oldhostinfo.relayState.CopyAllRelayFor()
  183. for _, r := range relayFor {
  184. existing, ok := newhostinfo.relayState.QueryRelayForByIp(r.PeerIp)
  185. var index uint32
  186. var relayFrom iputil.VpnIp
  187. var relayTo iputil.VpnIp
  188. switch {
  189. case ok && existing.State == Established:
  190. // This relay already exists in newhostinfo, then do nothing.
  191. continue
  192. case ok && existing.State == Requested:
  193. // The relay exists in a Requested state; re-send the request
  194. index = existing.LocalIndex
  195. switch r.Type {
  196. case TerminalType:
  197. relayFrom = newhostinfo.vpnIp
  198. relayTo = existing.PeerIp
  199. case ForwardingType:
  200. relayFrom = existing.PeerIp
  201. relayTo = newhostinfo.vpnIp
  202. default:
  203. // should never happen
  204. }
  205. case !ok:
  206. n.relayUsedLock.RLock()
  207. if _, relayUsed := n.relayUsed[r.LocalIndex]; !relayUsed {
  208. // The relay hasn't been used; don't migrate it.
  209. n.relayUsedLock.RUnlock()
  210. continue
  211. }
  212. n.relayUsedLock.RUnlock()
  213. // The relay doesn't exist at all; create some relay state and send the request.
  214. var err error
  215. index, err = AddRelay(n.l, newhostinfo, n.hostMap, r.PeerIp, nil, r.Type, Requested)
  216. if err != nil {
  217. n.l.WithError(err).Error("failed to migrate relay to new hostinfo")
  218. continue
  219. }
  220. switch r.Type {
  221. case TerminalType:
  222. relayFrom = newhostinfo.vpnIp
  223. relayTo = r.PeerIp
  224. case ForwardingType:
  225. relayFrom = r.PeerIp
  226. relayTo = newhostinfo.vpnIp
  227. default:
  228. // should never happen
  229. }
  230. }
  231. // Send a CreateRelayRequest to the peer.
  232. req := NebulaControl{
  233. Type: NebulaControl_CreateRelayRequest,
  234. InitiatorRelayIndex: index,
  235. RelayFromIp: uint32(relayFrom),
  236. RelayToIp: uint32(relayTo),
  237. }
  238. msg, err := req.Marshal()
  239. if err != nil {
  240. n.l.WithError(err).Error("failed to marshal Control message to migrate relay")
  241. } else {
  242. n.intf.SendMessageToHostInfo(header.Control, 0, newhostinfo, msg, make([]byte, 12), make([]byte, mtu))
  243. n.l.WithFields(logrus.Fields{
  244. "relayFrom": iputil.VpnIp(req.RelayFromIp),
  245. "relayTo": iputil.VpnIp(req.RelayToIp),
  246. "initiatorRelayIndex": req.InitiatorRelayIndex,
  247. "responderRelayIndex": req.ResponderRelayIndex,
  248. "vpnIp": newhostinfo.vpnIp}).
  249. Info("send CreateRelayRequest")
  250. }
  251. }
  252. }
  253. func (n *connectionManager) makeTrafficDecision(localIndex uint32, p, nb, out []byte, now time.Time) (trafficDecision, *HostInfo, *HostInfo) {
  254. n.hostMap.RLock()
  255. defer n.hostMap.RUnlock()
  256. hostinfo := n.hostMap.Indexes[localIndex]
  257. if hostinfo == nil {
  258. n.l.WithField("localIndex", localIndex).Debugf("Not found in hostmap")
  259. delete(n.pendingDeletion, localIndex)
  260. return doNothing, nil, nil
  261. }
  262. if n.isInvalidCertificate(now, hostinfo) {
  263. delete(n.pendingDeletion, hostinfo.localIndexId)
  264. return closeTunnel, hostinfo, nil
  265. }
  266. primary := n.hostMap.Hosts[hostinfo.vpnIp]
  267. mainHostInfo := true
  268. if primary != nil && primary != hostinfo {
  269. mainHostInfo = false
  270. }
  271. // Check for traffic on this hostinfo
  272. inTraffic, outTraffic := n.getAndResetTrafficCheck(localIndex)
  273. // A hostinfo is determined alive if there is incoming traffic
  274. if inTraffic {
  275. decision := doNothing
  276. if n.l.Level >= logrus.DebugLevel {
  277. hostinfo.logger(n.l).
  278. WithField("tunnelCheck", m{"state": "alive", "method": "passive"}).
  279. Debug("Tunnel status")
  280. }
  281. delete(n.pendingDeletion, hostinfo.localIndexId)
  282. if mainHostInfo {
  283. n.tryRehandshake(hostinfo)
  284. } else {
  285. if n.shouldSwapPrimary(hostinfo, primary) {
  286. decision = swapPrimary
  287. } else {
  288. // migrate the relays to the primary, if in use.
  289. decision = migrateRelays
  290. }
  291. }
  292. n.trafficTimer.Add(hostinfo.localIndexId, n.checkInterval)
  293. if !outTraffic {
  294. // Send a punch packet to keep the NAT state alive
  295. n.sendPunch(hostinfo)
  296. }
  297. return decision, hostinfo, primary
  298. }
  299. if _, ok := n.pendingDeletion[hostinfo.localIndexId]; ok {
  300. // We have already sent a test packet and nothing was returned, this hostinfo is dead
  301. hostinfo.logger(n.l).
  302. WithField("tunnelCheck", m{"state": "dead", "method": "active"}).
  303. Info("Tunnel status")
  304. delete(n.pendingDeletion, hostinfo.localIndexId)
  305. return deleteTunnel, hostinfo, nil
  306. }
  307. if hostinfo != nil && hostinfo.ConnectionState != nil && mainHostInfo {
  308. if !outTraffic {
  309. // If we aren't sending or receiving traffic then its an unused tunnel and we don't to test the tunnel.
  310. // Just maintain NAT state if configured to do so.
  311. n.sendPunch(hostinfo)
  312. n.trafficTimer.Add(hostinfo.localIndexId, n.checkInterval)
  313. return doNothing, nil, nil
  314. }
  315. if n.punchy.GetTargetEverything() {
  316. // This is similar to the old punchy behavior with a slight optimization.
  317. // We aren't receiving traffic but we are sending it, punch on all known
  318. // ips in case we need to re-prime NAT state
  319. n.sendPunch(hostinfo)
  320. }
  321. if n.l.Level >= logrus.DebugLevel {
  322. hostinfo.logger(n.l).
  323. WithField("tunnelCheck", m{"state": "testing", "method": "active"}).
  324. Debug("Tunnel status")
  325. }
  326. // Send a test packet to trigger an authenticated tunnel test, this should suss out any lingering tunnel issues
  327. n.intf.SendMessageToHostInfo(header.Test, header.TestRequest, hostinfo, p, nb, out)
  328. } else {
  329. if n.l.Level >= logrus.DebugLevel {
  330. hostinfo.logger(n.l).Debugf("Hostinfo sadness")
  331. }
  332. }
  333. n.pendingDeletion[hostinfo.localIndexId] = struct{}{}
  334. n.trafficTimer.Add(hostinfo.localIndexId, n.pendingDeletionInterval)
  335. return doNothing, nil, nil
  336. }
  337. func (n *connectionManager) shouldSwapPrimary(current, primary *HostInfo) bool {
  338. // The primary tunnel is the most recent handshake to complete locally and should work entirely fine.
  339. // If we are here then we have multiple tunnels for a host pair and neither side believes the same tunnel is primary.
  340. // Let's sort this out.
  341. if current.vpnIp < n.intf.myVpnIp {
  342. // Only one side should flip primary because if both flip then we may never resolve to a single tunnel.
  343. // vpn ip is static across all tunnels for this host pair so lets use that to determine who is flipping.
  344. // The remotes vpn ip is lower than mine. I will not flip.
  345. return false
  346. }
  347. certState := n.intf.certState.Load()
  348. return bytes.Equal(current.ConnectionState.certState.certificate.Signature, certState.certificate.Signature)
  349. }
  350. func (n *connectionManager) swapPrimary(current, primary *HostInfo) {
  351. n.hostMap.Lock()
  352. // Make sure the primary is still the same after the write lock. This avoids a race with a rehandshake.
  353. if n.hostMap.Hosts[current.vpnIp] == primary {
  354. n.hostMap.unlockedMakePrimary(current)
  355. }
  356. n.hostMap.Unlock()
  357. }
  358. // isInvalidCertificate will check if we should destroy a tunnel if pki.disconnect_invalid is true and
  359. // the certificate is no longer valid. Block listed certificates will skip the pki.disconnect_invalid
  360. // check and return true.
  361. func (n *connectionManager) isInvalidCertificate(now time.Time, hostinfo *HostInfo) bool {
  362. remoteCert := hostinfo.GetCert()
  363. if remoteCert == nil {
  364. return false
  365. }
  366. valid, err := remoteCert.Verify(now, n.intf.caPool)
  367. if valid {
  368. return false
  369. }
  370. if !n.intf.disconnectInvalid && err != cert.ErrBlockListed {
  371. // Block listed certificates should always be disconnected
  372. return false
  373. }
  374. fingerprint, _ := remoteCert.Sha256Sum()
  375. hostinfo.logger(n.l).WithError(err).
  376. WithField("fingerprint", fingerprint).
  377. Info("Remote certificate is no longer valid, tearing down the tunnel")
  378. return true
  379. }
  380. func (n *connectionManager) sendPunch(hostinfo *HostInfo) {
  381. if !n.punchy.GetPunch() {
  382. // Punching is disabled
  383. return
  384. }
  385. if n.punchy.GetTargetEverything() {
  386. hostinfo.remotes.ForEach(n.hostMap.preferredRanges, func(addr *udp.Addr, preferred bool) {
  387. n.metricsTxPunchy.Inc(1)
  388. n.intf.outside.WriteTo([]byte{1}, addr)
  389. })
  390. } else if hostinfo.remote != nil {
  391. n.metricsTxPunchy.Inc(1)
  392. n.intf.outside.WriteTo([]byte{1}, hostinfo.remote)
  393. }
  394. }
  395. func (n *connectionManager) tryRehandshake(hostinfo *HostInfo) {
  396. certState := n.intf.certState.Load()
  397. if bytes.Equal(hostinfo.ConnectionState.certState.certificate.Signature, certState.certificate.Signature) {
  398. return
  399. }
  400. n.l.WithField("vpnIp", hostinfo.vpnIp).
  401. WithField("reason", "local certificate is not current").
  402. Info("Re-handshaking with remote")
  403. //TODO: this is copied from getOrHandshake to keep the extra checks out of the hot path, figure it out
  404. newHostinfo := n.intf.handshakeManager.AddVpnIp(hostinfo.vpnIp, n.intf.initHostInfo)
  405. if !newHostinfo.HandshakeReady {
  406. ixHandshakeStage0(n.intf, newHostinfo.vpnIp, newHostinfo)
  407. }
  408. //If this is a static host, we don't need to wait for the HostQueryReply
  409. //We can trigger the handshake right now
  410. if _, ok := n.intf.lightHouse.GetStaticHostList()[hostinfo.vpnIp]; ok {
  411. select {
  412. case n.intf.handshakeManager.trigger <- hostinfo.vpnIp:
  413. default:
  414. }
  415. }
  416. }