mirror of
https://github.com/slackhq/nebula.git
synced 2026-08-15 19:36:59 +02:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 983784bc41 | |||
| 30e2735af3 | |||
| bce0b4249e |
@@ -105,11 +105,12 @@ func (cm *connectionManager) getInactivityTimeout() time.Duration {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (cm *connectionManager) In(h *HostInfo) {
|
func (cm *connectionManager) In(h *HostInfo) {
|
||||||
h.in.Store(true)
|
h.markIn()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cm *connectionManager) Out(h *HostInfo) {
|
// Out records outbound traffic and reports whether the local network changed since this tunnel last sent.
|
||||||
h.out.Store(true)
|
func (cm *connectionManager) Out(h *HostInfo) bool {
|
||||||
|
return h.markOut(cm.intf.rebindEpoch.Load())
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cm *connectionManager) RelayUsed(localIndex uint32) {
|
func (cm *connectionManager) RelayUsed(localIndex uint32) {
|
||||||
@@ -128,8 +129,7 @@ func (cm *connectionManager) RelayUsed(localIndex uint32) {
|
|||||||
// getAndResetTrafficCheck returns if there was any inbound or outbound traffic within the last tick and
|
// getAndResetTrafficCheck returns if there was any inbound or outbound traffic within the last tick and
|
||||||
// resets the state for this local index
|
// resets the state for this local index
|
||||||
func (cm *connectionManager) getAndResetTrafficCheck(h *HostInfo, now time.Time) (bool, bool) {
|
func (cm *connectionManager) getAndResetTrafficCheck(h *HostInfo, now time.Time) (bool, bool) {
|
||||||
in := h.in.Swap(false)
|
in, out := h.takeTraffic()
|
||||||
out := h.out.Swap(false)
|
|
||||||
if in || out {
|
if in || out {
|
||||||
h.lastUsed = now
|
h.lastUsed = now
|
||||||
}
|
}
|
||||||
@@ -340,7 +340,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim
|
|||||||
"tunnelCheck", m{"state": "alive", "method": "passive"},
|
"tunnelCheck", m{"state": "alive", "method": "passive"},
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
hostinfo.pendingDeletion.Store(false)
|
hostinfo.setPendingDeletion(false)
|
||||||
|
|
||||||
if mainHostInfo {
|
if mainHostInfo {
|
||||||
decision = tryRehandshake
|
decision = tryRehandshake
|
||||||
@@ -363,7 +363,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim
|
|||||||
return decision, hostinfo, primary
|
return decision, hostinfo, primary
|
||||||
}
|
}
|
||||||
|
|
||||||
if hostinfo.pendingDeletion.Load() {
|
if hostinfo.isPendingDeletion() {
|
||||||
// We have already sent a test packet and nothing was returned, this hostinfo is dead
|
// We have already sent a test packet and nothing was returned, this hostinfo is dead
|
||||||
hostinfo.logger(cm.l).Info("Tunnel status",
|
hostinfo.logger(cm.l).Info("Tunnel status",
|
||||||
"tunnelCheck", m{"state": "dead", "method": "active"},
|
"tunnelCheck", m{"state": "dead", "method": "active"},
|
||||||
@@ -414,7 +414,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
hostinfo.pendingDeletion.Store(true)
|
hostinfo.setPendingDeletion(true)
|
||||||
cm.trafficTimer.Add(hostinfo.localIndexId, cm.pendingDeletionInterval)
|
cm.trafficTimer.Add(hostinfo.localIndexId, cm.pendingDeletionInterval)
|
||||||
return decision, hostinfo, nil
|
return decision, hostinfo, nil
|
||||||
}
|
}
|
||||||
|
|||||||
+36
-36
@@ -86,25 +86,25 @@ func Test_NewConnectionManagerTest(t *testing.T) {
|
|||||||
// We saw traffic out to vpnIp
|
// We saw traffic out to vpnIp
|
||||||
nc.Out(hostinfo)
|
nc.Out(hostinfo)
|
||||||
nc.In(hostinfo)
|
nc.In(hostinfo)
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.True(t, hostinfo.out.Load())
|
assert.True(t, hostinfo.sentSinceCheck())
|
||||||
assert.True(t, hostinfo.in.Load())
|
assert.True(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
// Do a traffic check tick, should not be pending deletion but should not have any in/out packets recorded
|
// Do a traffic check tick, should not be pending deletion but should not have any in/out packets recorded
|
||||||
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
// Do another traffic check tick, this host should be pending deletion now
|
// Do another traffic check tick, this host should be pending deletion now
|
||||||
nc.Out(hostinfo)
|
nc.Out(hostinfo)
|
||||||
assert.True(t, hostinfo.out.Load())
|
assert.True(t, hostinfo.sentSinceCheck())
|
||||||
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
||||||
assert.True(t, hostinfo.pendingDeletion.Load())
|
assert.True(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
|
|
||||||
@@ -168,33 +168,33 @@ func Test_NewConnectionManagerTest2(t *testing.T) {
|
|||||||
// We saw traffic out to vpnIp
|
// We saw traffic out to vpnIp
|
||||||
nc.Out(hostinfo)
|
nc.Out(hostinfo)
|
||||||
nc.In(hostinfo)
|
nc.In(hostinfo)
|
||||||
assert.True(t, hostinfo.in.Load())
|
assert.True(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.True(t, hostinfo.out.Load())
|
assert.True(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
|
|
||||||
// Do a traffic check tick, should not be pending deletion but should not have any in/out packets recorded
|
// Do a traffic check tick, should not be pending deletion but should not have any in/out packets recorded
|
||||||
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
// Do another traffic check tick, this host should be pending deletion now
|
// Do another traffic check tick, this host should be pending deletion now
|
||||||
nc.Out(hostinfo)
|
nc.Out(hostinfo)
|
||||||
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
||||||
assert.True(t, hostinfo.pendingDeletion.Load())
|
assert.True(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
|
|
||||||
// We saw traffic, should no longer be pending deletion
|
// We saw traffic, should no longer be pending deletion
|
||||||
nc.In(hostinfo)
|
nc.In(hostinfo)
|
||||||
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
}
|
}
|
||||||
@@ -253,31 +253,31 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) {
|
|||||||
// Do a traffic check tick, in and out should be cleared but should not be pending deletion
|
// Do a traffic check tick, in and out should be cleared but should not be pending deletion
|
||||||
nc.Out(hostinfo)
|
nc.Out(hostinfo)
|
||||||
nc.In(hostinfo)
|
nc.In(hostinfo)
|
||||||
assert.True(t, hostinfo.out.Load())
|
assert.True(t, hostinfo.sentSinceCheck())
|
||||||
assert.True(t, hostinfo.in.Load())
|
assert.True(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
decision, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, now)
|
decision, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, now)
|
||||||
assert.Equal(t, tryRehandshake, decision)
|
assert.Equal(t, tryRehandshake, decision)
|
||||||
assert.Equal(t, now, hostinfo.lastUsed)
|
assert.Equal(t, now, hostinfo.lastUsed)
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*5))
|
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*5))
|
||||||
assert.Equal(t, doNothing, decision)
|
assert.Equal(t, doNothing, decision)
|
||||||
assert.Equal(t, now, hostinfo.lastUsed)
|
assert.Equal(t, now, hostinfo.lastUsed)
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
|
|
||||||
// Do another traffic check tick, should still not be pending deletion
|
// Do another traffic check tick, should still not be pending deletion
|
||||||
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*10))
|
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*10))
|
||||||
assert.Equal(t, doNothing, decision)
|
assert.Equal(t, doNothing, decision)
|
||||||
assert.Equal(t, now, hostinfo.lastUsed)
|
assert.Equal(t, now, hostinfo.lastUsed)
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
|
|
||||||
@@ -285,9 +285,9 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) {
|
|||||||
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Minute*10))
|
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Minute*10))
|
||||||
assert.Equal(t, closeTunnel, decision)
|
assert.Equal(t, closeTunnel, decision)
|
||||||
assert.Equal(t, now, hostinfo.lastUsed)
|
assert.Equal(t, now, hostinfo.lastUsed)
|
||||||
assert.False(t, hostinfo.pendingDeletion.Load())
|
assert.False(t, hostinfo.isPendingDeletion())
|
||||||
assert.False(t, hostinfo.out.Load())
|
assert.False(t, hostinfo.sentSinceCheck())
|
||||||
assert.False(t, hostinfo.in.Load())
|
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
|
||||||
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
|
||||||
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -212,7 +212,7 @@ func (c *Control) RebindUDPServer() {
|
|||||||
c.f.lightHouse.SendUpdate()
|
c.f.lightHouse.SendUpdate()
|
||||||
|
|
||||||
// Let the main interface know that we rebound so that underlying tunnels know to trigger punches from their remotes
|
// Let the main interface know that we rebound so that underlying tunnels know to trigger punches from their remotes
|
||||||
c.f.rebindCount++
|
c.f.rebindEpoch.Add(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ListHostmapHosts returns details about the actual or pending (handshaking) hostmap by vpn ip
|
// ListHostmapHosts returns details about the actual or pending (handshaking) hostmap by vpn ip
|
||||||
|
|||||||
@@ -0,0 +1,136 @@
|
|||||||
|
//go:build e2e_testing
|
||||||
|
// +build e2e_testing
|
||||||
|
|
||||||
|
package e2e
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/slackhq/nebula"
|
||||||
|
"github.com/slackhq/nebula/cert"
|
||||||
|
"github.com/slackhq/nebula/cert_test"
|
||||||
|
"github.com/slackhq/nebula/e2e/router"
|
||||||
|
"github.com/slackhq/nebula/udp"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestRecoveryTiming measures how long a tunnel takes to come back after the peer stops accepting our traffic,
|
||||||
|
// which is what a laptop waking on a new network looks like from the peer's side: its NAT has no state for where
|
||||||
|
// we are now, so everything we send disappears.
|
||||||
|
//
|
||||||
|
// It is a measurement, not a pass/fail assertion. Recovery is timed to the moment the peer punches back at us,
|
||||||
|
// since that is when its NAT opens and the tunnel is usable again.
|
||||||
|
//
|
||||||
|
// go test -tags e2e_testing -v -run TestRecoveryTiming ./e2e/
|
||||||
|
func TestRecoveryTiming(t *testing.T) {
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
rebind bool
|
||||||
|
}{
|
||||||
|
{"no trigger", false},
|
||||||
|
{"rebind counter", true},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
d, lost := measureRecovery(t, tc.rebind)
|
||||||
|
t.Logf("RESULT %-16s recovered in %-9v (%d packets lost)", tc.name, d.Round(time.Millisecond), lost)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// measureRecovery returns how long until the peer punched back, and how many of our packets died meanwhile. When
|
||||||
|
// rebind is true we call RebindUDPServer once the tunnel goes dark, which is what the darwin network change
|
||||||
|
// monitor does and what iOS has always done. When false, nothing tells nebula anything is wrong.
|
||||||
|
func measureRecovery(t *testing.T, rebind bool) (time.Duration, int) {
|
||||||
|
t.Helper()
|
||||||
|
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
|
||||||
|
|
||||||
|
lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{
|
||||||
|
"lighthouse": m{"am_lighthouse": true},
|
||||||
|
})
|
||||||
|
|
||||||
|
peerCfg := m{
|
||||||
|
"lighthouse": m{
|
||||||
|
"hosts": []any{lhVpnIpNet[0].Addr().String()},
|
||||||
|
"interval": 600,
|
||||||
|
"local_allow_list": m{
|
||||||
|
"10.0.0.0/24": true,
|
||||||
|
"::/0": false,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"static_host_map": m{
|
||||||
|
lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", peerCfg)
|
||||||
|
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.3/24", peerCfg)
|
||||||
|
|
||||||
|
r := router.NewR(t, lhControl, myControl, theirControl)
|
||||||
|
defer r.RenderFlow()
|
||||||
|
defer func() {
|
||||||
|
lhControl.Stop()
|
||||||
|
myControl.Stop()
|
||||||
|
theirControl.Stop()
|
||||||
|
}()
|
||||||
|
|
||||||
|
lhControl.Start()
|
||||||
|
myControl.Start()
|
||||||
|
theirControl.Start()
|
||||||
|
r.RouteFor(time.Millisecond * 500)
|
||||||
|
|
||||||
|
myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
|
||||||
|
theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
|
||||||
|
|
||||||
|
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("establish")))
|
||||||
|
r.RouteFor(time.Second)
|
||||||
|
if myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false) == nil {
|
||||||
|
t.Fatal("failed to establish the tunnel we are measuring")
|
||||||
|
}
|
||||||
|
r.RouteFor(time.Millisecond * 500)
|
||||||
|
|
||||||
|
// From here the peer's NAT has no state for us, everything we send it disappears
|
||||||
|
start := time.Now()
|
||||||
|
blackholed := 0
|
||||||
|
var recovered time.Duration
|
||||||
|
|
||||||
|
if rebind {
|
||||||
|
myControl.RebindUDPServer()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Keep the tun busy the way someone retrying a stalled connection would
|
||||||
|
stop := make(chan struct{})
|
||||||
|
defer close(stop)
|
||||||
|
go func() {
|
||||||
|
tick := time.NewTicker(time.Millisecond * 200)
|
||||||
|
defer tick.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-stop:
|
||||||
|
return
|
||||||
|
case <-tick.C:
|
||||||
|
myControl.InjectTunPacket(BuildTunUDPPacket(
|
||||||
|
theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("retry")))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
r.RouteForAllExitFuncOrTimeout(time.Second*30, func(p *udp.Packet, c *nebula.Control) router.ExitType {
|
||||||
|
if c == theirControl && p.From == myControl.GetUDPAddr() {
|
||||||
|
blackholed++
|
||||||
|
return router.Drop
|
||||||
|
}
|
||||||
|
|
||||||
|
// The peer reaching us directly is the moment its NAT opened, whether that is a punch or a handshake
|
||||||
|
if c == myControl && p.From == theirUdpAddr {
|
||||||
|
recovered = time.Since(start)
|
||||||
|
return router.RouteAndExit
|
||||||
|
}
|
||||||
|
|
||||||
|
return router.KeepRouting
|
||||||
|
})
|
||||||
|
|
||||||
|
if recovered == 0 {
|
||||||
|
t.Fatalf("no recovery within 30s (%d packets blackholed)", blackholed)
|
||||||
|
}
|
||||||
|
return recovered, blackholed
|
||||||
|
}
|
||||||
+19
-2
@@ -153,6 +153,9 @@ const (
|
|||||||
ExitNow ExitType = 1
|
ExitNow ExitType = 1
|
||||||
// RouteAndExit routes this packet and exits immediately afterwards
|
// RouteAndExit routes this packet and exits immediately afterwards
|
||||||
RouteAndExit ExitType = 2
|
RouteAndExit ExitType = 2
|
||||||
|
// Drop discards this packet without delivering it and keeps routing. Use it to simulate a blackhole, such as
|
||||||
|
// a restrictive NAT refusing traffic from an address it has not seen.
|
||||||
|
Drop ExitType = 3
|
||||||
)
|
)
|
||||||
|
|
||||||
type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
|
type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
|
||||||
@@ -163,7 +166,9 @@ type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
|
|||||||
func NewR(t testing.TB, controls ...*nebula.Control) *R {
|
func NewR(t testing.TB, controls ...*nebula.Control) *R {
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
|
||||||
if err := os.MkdirAll("mermaid", 0755); err != nil {
|
// t.Name() contains a slash for subtests, so the flow log can land in a nested directory
|
||||||
|
fn := filepath.Join("mermaid", fmt.Sprintf("%s.md", t.Name()))
|
||||||
|
if err := os.MkdirAll(filepath.Dir(fn), 0755); err != nil {
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -174,7 +179,7 @@ func NewR(t testing.TB, controls ...*nebula.Control) *R {
|
|||||||
outNat: make(map[outNatKey]netip.AddrPort),
|
outNat: make(map[outNatKey]netip.AddrPort),
|
||||||
flow: []flowEntry{},
|
flow: []flowEntry{},
|
||||||
ignoreFlows: []ignoreFlow{},
|
ignoreFlows: []ignoreFlow{},
|
||||||
fn: filepath.Join("mermaid", fmt.Sprintf("%s.md", t.Name())),
|
fn: fn,
|
||||||
t: t,
|
t: t,
|
||||||
cancelRender: cancel,
|
cancelRender: cancel,
|
||||||
}
|
}
|
||||||
@@ -687,6 +692,10 @@ func (r *R) RouteExitFunc(sender *nebula.Control, whatDo ExitFunc) {
|
|||||||
p.Release()
|
p.Release()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
case Drop:
|
||||||
|
// Record it so the flow log shows the attempt, but never hand it to the receiver
|
||||||
|
r.unlockedInjectFlow(sender, receiver, p, false)
|
||||||
|
|
||||||
case KeepRouting:
|
case KeepRouting:
|
||||||
fp := r.unlockedInjectFlow(sender, receiver, p, false)
|
fp := r.unlockedInjectFlow(sender, receiver, p, false)
|
||||||
receiver.InjectUDPPacket(p)
|
receiver.InjectUDPPacket(p)
|
||||||
@@ -779,6 +788,10 @@ func (r *R) RouteForAllExitFuncOrTimeout(timeout time.Duration, whatDo ExitFunc)
|
|||||||
p.Release()
|
p.Release()
|
||||||
return true
|
return true
|
||||||
|
|
||||||
|
case Drop:
|
||||||
|
// Record it so the flow log shows the attempt, but never hand it to the receiver
|
||||||
|
r.unlockedInjectFlow(cm[x], receiver, p, false)
|
||||||
|
|
||||||
case KeepRouting:
|
case KeepRouting:
|
||||||
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
|
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
|
||||||
receiver.InjectUDPPacket(p)
|
receiver.InjectUDPPacket(p)
|
||||||
@@ -884,6 +897,10 @@ func (r *R) RouteForAllExitFunc(whatDo ExitFunc) {
|
|||||||
p.Release()
|
p.Release()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
case Drop:
|
||||||
|
// Record it so the flow log shows the attempt, but never hand it to the receiver
|
||||||
|
r.unlockedInjectFlow(cm[x], receiver, p, false)
|
||||||
|
|
||||||
case KeepRouting:
|
case KeepRouting:
|
||||||
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
|
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
|
||||||
receiver.InjectUDPPacket(p)
|
receiver.InjectUDPPacket(p)
|
||||||
|
|||||||
+73
-11
@@ -238,18 +238,30 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type HostInfo struct {
|
type HostInfo struct {
|
||||||
|
// The first cache line is everything the packet paths touch. Grouping them here means a send or receive
|
||||||
|
// pulls in one line instead of two, which is what the layout looked like when state lived at the end.
|
||||||
|
|
||||||
remote atomic.Pointer[netip.AddrPort]
|
remote atomic.Pointer[netip.AddrPort]
|
||||||
remotes *RemoteList
|
|
||||||
promoteCounter atomic.Uint32
|
|
||||||
ConnectionState *ConnectionState
|
ConnectionState *ConnectionState
|
||||||
|
|
||||||
|
// state holds everything the hot paths need to touch per packet, in one word: whether we have seen traffic
|
||||||
|
// each way since the connection manager last looked, whether it has given up on us, and the
|
||||||
|
// Interface.rebindEpoch this tunnel last sent under. Keeping the epoch here means it survives the traffic
|
||||||
|
// bits being cleared, so a tunnel that has not sent since a rebind still notices when it does.
|
||||||
|
state atomic.Uint32
|
||||||
|
|
||||||
|
promoteCounter atomic.Uint32
|
||||||
remoteIndexId uint32
|
remoteIndexId uint32
|
||||||
localIndexId uint32
|
localIndexId uint32
|
||||||
|
remotes *RemoteList
|
||||||
|
|
||||||
// vpnAddrs is a list of vpn addresses assigned to this host that are within our own vpn networks
|
// vpnAddrs is a list of vpn addresses assigned to this host that are within our own vpn networks
|
||||||
// The host may have other vpn addresses that are outside our
|
// The host may have other vpn addresses that are outside our
|
||||||
// vpn networks but were removed because they are not usable
|
// vpn networks but were removed because they are not usable
|
||||||
vpnAddrs []netip.Addr
|
vpnAddrs []netip.Addr
|
||||||
|
|
||||||
|
// Everything below is off the packet path: handshakes, relays, roaming and the connection manager.
|
||||||
|
|
||||||
// networks is a combination of specific vpn addresses (not prefixes!) and full unsafe networks assigned to this host.
|
// networks is a combination of specific vpn addresses (not prefixes!) and full unsafe networks assigned to this host.
|
||||||
networks *bart.Table[NetworkType]
|
networks *bart.Table[NetworkType]
|
||||||
relayState RelayState
|
relayState RelayState
|
||||||
@@ -262,11 +274,6 @@ type HostInfo struct {
|
|||||||
// This is used to limit lighthouse re-queries in chatty clients
|
// This is used to limit lighthouse re-queries in chatty clients
|
||||||
nextLHQuery atomic.Int64
|
nextLHQuery atomic.Int64
|
||||||
|
|
||||||
// lastRebindCount is the other side of Interface.rebindCount, if these values don't match then we need to ask LH
|
|
||||||
// for a punch from the remote end of this tunnel. The goal being to prime their conntrack for our traffic just like
|
|
||||||
// with a handshake
|
|
||||||
lastRebindCount int8
|
|
||||||
|
|
||||||
// lastHandshakeTime records the time the remote side told us about at the stage when the handshake was completed locally
|
// lastHandshakeTime records the time the remote side told us about at the stage when the handshake was completed locally
|
||||||
// Stage 1 packet will contain it if I am a responder, stage 2 packet if I am an initiator
|
// Stage 1 packet will contain it if I am a responder, stage 2 packet if I am an initiator
|
||||||
// This is used to avoid an attack where a handshake packet is replayed after some time
|
// This is used to avoid an attack where a handshake packet is replayed after some time
|
||||||
@@ -275,9 +282,6 @@ type HostInfo struct {
|
|||||||
lastRoam time.Time
|
lastRoam time.Time
|
||||||
lastRoamRemote netip.AddrPort
|
lastRoamRemote netip.AddrPort
|
||||||
|
|
||||||
//TODO: in, out, and others might benefit from being an atomic.Int32. We could collapse connectionManager pendingDeletion, relayUsed, and in/out into this 1 thing
|
|
||||||
in, out, pendingDeletion atomic.Bool
|
|
||||||
|
|
||||||
// lastUsed tracks the last time ConnectionManager checked the tunnel and it was in use.
|
// lastUsed tracks the last time ConnectionManager checked the tunnel and it was in use.
|
||||||
// This value will be behind against actual tunnel utilization in the hot path.
|
// This value will be behind against actual tunnel utilization in the hot path.
|
||||||
// This should only be used by the ConnectionManagers ticker routine.
|
// This should only be used by the ConnectionManagers ticker routine.
|
||||||
@@ -658,7 +662,7 @@ func (hm *HostMap) unlockedAddHostInfo(hostinfo *HostInfo, f *Interface) {
|
|||||||
hm.Indexes[hostinfo.localIndexId] = hostinfo
|
hm.Indexes[hostinfo.localIndexId] = hostinfo
|
||||||
hm.RemoteIndexes[hostinfo.remoteIndexId] = hostinfo
|
hm.RemoteIndexes[hostinfo.remoteIndexId] = hostinfo
|
||||||
|
|
||||||
hostinfo.out.Store(true)
|
hostinfo.markOut(f.rebindEpoch.Load())
|
||||||
if f.connectionManager != nil { // f.connectionManager is only nil in some unit tests
|
if f.connectionManager != nil { // f.connectionManager is only nil in some unit tests
|
||||||
f.connectionManager.trafficTimer.Add(hostinfo.localIndexId, f.connectionManager.checkInterval)
|
f.connectionManager.trafficTimer.Add(hostinfo.localIndexId, f.connectionManager.checkInterval)
|
||||||
}
|
}
|
||||||
@@ -759,6 +763,64 @@ func (i *HostInfo) TryPromoteBest(preferredRanges []netip.Prefix, ifce *Interfac
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Bits within HostInfo.state. Everything above stateEpochShift is the rebind epoch.
|
||||||
|
const (
|
||||||
|
stateIn uint32 = 1 << iota
|
||||||
|
stateOut
|
||||||
|
statePendingDeletion
|
||||||
|
|
||||||
|
stateFlags = stateIn | stateOut | statePendingDeletion
|
||||||
|
stateEpochShift = 3
|
||||||
|
)
|
||||||
|
|
||||||
|
// markIn records inbound traffic. Reading first keeps the cache line shared on the common path, where the bit
|
||||||
|
// is already set.
|
||||||
|
func (i *HostInfo) markIn() {
|
||||||
|
if i.state.Load()&stateIn == 0 {
|
||||||
|
i.state.Or(stateIn)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// markOut records that we sent on this tunnel under the given rebind epoch. It reports whether the epoch moved
|
||||||
|
// since our last send, which means the local network changed and we want the far side to punch at us again.
|
||||||
|
// The common path is a single load that matches and returns.
|
||||||
|
func (i *HostInfo) markOut(epoch uint32) bool {
|
||||||
|
e := epoch << stateEpochShift
|
||||||
|
for {
|
||||||
|
old := i.state.Load()
|
||||||
|
if old&stateOut != 0 && old&^stateFlags == e {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
if i.state.CompareAndSwap(old, old&stateFlags|stateOut|e) {
|
||||||
|
return old&^stateFlags != e
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// sentSinceCheck reports whether anything has been sent since the connection manager last looked.
|
||||||
|
func (i *HostInfo) sentSinceCheck() bool {
|
||||||
|
return i.state.Load()&stateOut != 0
|
||||||
|
}
|
||||||
|
|
||||||
|
// takeTraffic clears both traffic bits, leaving the epoch alone, and reports what they were.
|
||||||
|
func (i *HostInfo) takeTraffic() (in bool, out bool) {
|
||||||
|
old := i.state.And(^(stateIn | stateOut))
|
||||||
|
return old&stateIn != 0, old&stateOut != 0
|
||||||
|
}
|
||||||
|
|
||||||
|
func (i *HostInfo) setPendingDeletion(v bool) {
|
||||||
|
if v {
|
||||||
|
i.state.Or(statePendingDeletion)
|
||||||
|
} else {
|
||||||
|
i.state.And(^statePendingDeletion)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (i *HostInfo) isPendingDeletion() bool {
|
||||||
|
return i.state.Load()&statePendingDeletion != 0
|
||||||
|
}
|
||||||
|
|
||||||
func (i *HostInfo) GetCert() *cert.CachedCertificate {
|
func (i *HostInfo) GetCert() *cert.CachedCertificate {
|
||||||
if i.ConnectionState != nil {
|
if i.ConnectionState != nil {
|
||||||
return i.ConnectionState.peerCert
|
return i.ConnectionState.peerCert
|
||||||
|
|||||||
@@ -365,17 +365,12 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
|
|||||||
|
|
||||||
//l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p)
|
//l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p)
|
||||||
out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
|
out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
|
||||||
f.connectionManager.Out(hostinfo)
|
// We rebound since this tunnel last sent, so the local network moved. Ask the lighthouse to have the far side
|
||||||
|
// punch at where we are now, which primes their conntrack the same way a handshake would.
|
||||||
// Query our LH if we haven't since the last time we've been rebound, this will cause the remote to punch against
|
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
|
||||||
// all our addrs and enable a faster roaming.
|
|
||||||
if t != header.CloseTunnel && hostinfo.lastRebindCount != f.rebindCount {
|
|
||||||
//NOTE: there is an update hole if a tunnel isn't used and exactly 256 rebinds occur before the tunnel is
|
|
||||||
// finally used again. This tunnel would eventually be torn down and recreated if this action didn't help.
|
|
||||||
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
||||||
hostinfo.lastRebindCount = f.rebindCount
|
|
||||||
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
||||||
f.l.Debug("Lighthouse update triggered for punch due to rebind counter",
|
f.l.Debug("Lighthouse update triggered for punch due to rebind epoch",
|
||||||
"vpnAddrs", hostinfo.vpnAddrs,
|
"vpnAddrs", hostinfo.vpnAddrs,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-2
@@ -82,8 +82,10 @@ type Interface struct {
|
|||||||
sendRecvErrorConfig recvErrorConfig
|
sendRecvErrorConfig recvErrorConfig
|
||||||
acceptRecvErrorConfig recvErrorConfig
|
acceptRecvErrorConfig recvErrorConfig
|
||||||
|
|
||||||
// rebindCount is used to decide if an active tunnel should trigger a punch notification through a lighthouse
|
// rebindEpoch bumps every time the udp listener is rebound, which means the local network moved. Tunnels
|
||||||
rebindCount int8
|
// compare it against their own copy to decide they need a punch from the far side. Read on every send, only
|
||||||
|
// written on a rebind, so the cache line stays shared across the routines.
|
||||||
|
rebindEpoch atomic.Uint32
|
||||||
version string
|
version string
|
||||||
|
|
||||||
conntrackCacheTimeout time.Duration
|
conntrackCacheTimeout time.Duration
|
||||||
|
|||||||
Reference in New Issue
Block a user