Compare commits

..

2 Commits

Author SHA1 Message Date
Nate Brown 35ac12b0c0 Group the HostInfo fields the packet paths touch 2026-07-23 21:31:51 -05:00
Nate Brown c07f28cd04 Fold the rebind counter and traffic flags into one atomic word 2026-07-23 20:36:51 -05:00
12 changed files with 240 additions and 218 deletions
+13 -8
View File
@@ -105,11 +105,17 @@ 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) { // OutRelay records relayed traffic, leaving the rebind epoch for the direct path to this host to consume
h.out.Store(true) func (cm *connectionManager) OutRelay(h *HostInfo) {
h.markOutOnly()
}
// Out records outbound traffic and reports whether we rebound since this tunnel last sent
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 +134,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 +345,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 +368,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 +419,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
View File
@@ -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
View File
@@ -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
+10
View File
@@ -123,6 +123,16 @@ func (c *Control) SetLocalAddrsFn(fn func(*LocalAllowList) []netip.Addr) {
c.f.lightHouse.localAddrsFn = fn c.f.lightHouse.localAddrsFn = fn
} }
// GetRebindEpochFor returns the rebind epoch a tunnel last sent under, so a test can tell whether a send
// consumed the epoch edge without having to infer it from lighthouse traffic.
func (c *Control) GetRebindEpochFor(vpnAddr netip.Addr) (uint32, bool) {
h := c.f.hostMap.QueryVpnAddr(vpnAddr)
if h == nil {
return 0, false
}
return h.state.Load() >> stateEpochShift, true
}
func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool { func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool {
hostinfo := c.f.handshakeManager.QueryVpnAddr(vpnIp) hostinfo := c.f.handshakeManager.QueryVpnAddr(vpnIp)
if hostinfo == nil { if hostinfo == nil {
+54
View File
@@ -223,3 +223,57 @@ func TestRebindAdvertisesNewAddressAfterMove(t *testing.T) {
lhControl.Stop() lhControl.Stop()
myControl.Stop() myControl.Stop()
} }
// A relayed send records traffic but must not consume the rebind epoch. If it does, the next direct send to the
// relay host sees the epoch already current and never requeries, so the far side is never told to punch at our
// new address. This pins the SendVia call site, which the unit tests cannot reach.
func TestRebindRequeriesAfterRelayedSend(t *testing.T) {
t.Parallel()
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
// No lighthouse on purpose: it would hand out a direct address for them and nothing would relay.
myControl, myVpnIpNet, _, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.1/24", m{"relay": m{"use_relays": true}})
relayControl, relayVpnIpNet, relayUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "relay", "10.128.0.128/24", m{"relay": m{"am_relay": true}})
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.2/24", m{"relay": m{"use_relays": true}})
myControl.InjectLightHouseAddr(relayVpnIpNet[0].Addr(), relayUdpAddr)
myControl.InjectRelays(theirVpnIpNet[0].Addr(), []netip.Addr{relayVpnIpNet[0].Addr()})
relayControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
r := router.NewR(t, myControl, relayControl, theirControl)
defer r.RenderFlow()
myControl.Start()
relayControl.Start()
theirControl.Start()
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("establish")))
r.RouteForAllUntilTxTun(theirControl)
r.RouteFor(time.Millisecond * 500)
hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
require.NotNil(t, hi, "expected a tunnel to them")
require.NotEmpty(t, hi.CurrentRelaysToMe, "them must be reachable only via the relay for this test to mean anything")
// sendNoMetrics only reaches SendVia when there is no direct remote, so pin that too. Without this the test
// keeps passing while quietly sending direct and never exercising the relay path.
require.False(t, hi.CurrentRemote.IsValid(), "them must have no direct remote, otherwise SendVia is never called")
before, ok := myControl.GetRebindEpochFor(relayVpnIpNet[0].Addr())
require.True(t, ok, "expected a tunnel to the relay")
myControl.RebindUDPServer()
// Traffic to them goes through SendVia on the relay tunnel. That must record traffic without consuming the
// relay tunnel's own epoch edge, which belongs to the direct path.
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("relayed")))
r.RouteForAllUntilTxTun(theirControl)
after, ok := myControl.GetRebindEpochFor(relayVpnIpNet[0].Addr())
require.True(t, ok)
assert.Equal(t, before, after,
"a relayed send consumed the relay tunnel's rebind epoch, so the next direct send will not requery")
myControl.Stop()
relayControl.Stop()
theirControl.Stop()
}
+73 -11
View File
@@ -238,18 +238,26 @@ const (
) )
type HostInfo struct { type HostInfo struct {
// The first cache line is everything the packet paths touch.
remote atomic.Pointer[netip.AddrPort] remote atomic.Pointer[netip.AddrPort]
remotes *RemoteList
promoteCounter atomic.Uint32
ConnectionState *ConnectionState ConnectionState *ConnectionState
// Traffic bits, pendingDeletion, and the rebind epoch we last sent under
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 +270,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 +278,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 +658,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 +759,68 @@ func (i *HostInfo) TryPromoteBest(preferredRanges []netip.Prefix, ifce *Interfac
} }
} }
// Bits within HostInfo.state, everything above stateEpochShift is the epoch
const (
stateIn uint32 = 1 << iota
stateOut
statePendingDeletion
stateFlags = stateIn | stateOut | statePendingDeletion
stateEpochShift = 3
)
// markIn records inbound traffic
func (i *HostInfo) markIn() {
if i.state.Load()&stateIn == 0 {
i.state.Or(stateIn)
}
}
// markOut records a send and reports whether the epoch moved, meaning we want a punch from the far side
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
}
}
}
// markOutOnly records a send without consuming the rebind epoch, for paths that cannot act on a requery
func (i *HostInfo) markOutOnly() {
if i.state.Load()&stateOut == 0 {
i.state.Or(stateOut)
}
}
// 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
+40
View File
@@ -401,3 +401,43 @@ func TestHostMap_RelayState(t *testing.T) {
assert.Equal(t, []netip.Addr{}, h1.relayState.relays) assert.Equal(t, []netip.Addr{}, h1.relayState.relays)
} }
func TestHostInfo_markOut(t *testing.T) {
h := &HostInfo{}
h.markOut(5) // stamped when the tunnel was added
// A tunnel already on the current epoch has nothing to report, which is what keeps a fresh tunnel from
// requerying on its first packet
assert.False(t, h.markOut(5), "an unchanged epoch should not report a move")
assert.True(t, h.sentSinceCheck(), "the send is still recorded as traffic")
// A rebind is observed exactly once, so we requery once per rebind
assert.True(t, h.markOut(6), "a bumped epoch should report a move")
assert.False(t, h.markOut(6), "the epoch move should only be reported once")
// Traffic and pendingDeletion live in the same word and must survive an epoch change
h.setPendingDeletion(true)
h.markIn()
assert.True(t, h.markOut(7))
assert.True(t, h.isPendingDeletion(), "pendingDeletion must survive an epoch change")
in, out := h.takeTraffic()
assert.True(t, in, "inbound traffic must survive an epoch change")
assert.True(t, out)
// Clearing the traffic bits leaves the epoch alone, otherwise an idle tunnel would requery forever
assert.False(t, h.markOut(7), "takeTraffic must not disturb the epoch")
}
// A relayed send records traffic but must leave the rebind epoch for the direct path to consume, otherwise
// relaying to a host swallows the requery that gets the far side punching at our new address.
func TestHostInfo_markOutOnly(t *testing.T) {
h := &HostInfo{}
h.markOut(5)
h.markOutOnly()
assert.True(t, h.sentSinceCheck(), "a relayed send is still outbound traffic")
assert.False(t, h.markOut(5), "a relayed send must not disturb the epoch")
assert.True(t, h.markOut(6), "a relayed send must not consume the epoch edge")
assert.False(t, h.markOut(6))
}
+4 -10
View File
@@ -297,7 +297,7 @@ func (f *Interface) SendVia(via *HostInfo,
c := via.ConnectionState.messageCounter.Add(1) c := via.ConnectionState.messageCounter.Add(1)
out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c) out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c)
f.connectionManager.Out(via) f.connectionManager.OutRelay(via)
// Authenticate the header and payload, but do not encrypt for this message type. // Authenticate the header and payload, but do not encrypt for this message type.
// The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload. // The payload consists of the inner, unencrypted Nebula header, as well as the end-to-end encrypted payload.
@@ -365,17 +365,11 @@ 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, ask the lighthouse to get the far side punching at us again
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
// Query our LH if we haven't since the last time we've been rebound, this will cause the remote to punch against
// 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,
) )
} }
+2 -2
View File
@@ -82,8 +82,8 @@ 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 // Bumped on every udp rebind, tunnels compare it to decide they need a punch from the far side
rebindCount int8 rebindEpoch atomic.Uint32
version string version string
conntrackCacheTimeout time.Duration conntrackCacheTimeout time.Duration
+2 -5
View File
@@ -22,12 +22,9 @@ type m = map[string]any
func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) { func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
// The goroutines started below stop only when this context does, and only a caller holding the // Automatically cancel the context if Main returns an error, to signal all created goroutines to quit.
// Control can arrange that. Cancel whenever we are not handing one back, which covers an error
// and a config test alike: a config test used to leave the lighthouse query worker, and a
// hostname resolver per dns named static host, running for the life of the process.
defer func() { defer func() {
if retcon == nil { if reterr != nil {
cancel() cancel()
} }
}() }()
-82
View File
@@ -1,82 +0,0 @@
package nebula
import (
"fmt"
"net/netip"
"os"
"path/filepath"
"testing"
"time"
"github.com/slackhq/nebula/cert"
cert_test "github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/require"
"go.uber.org/goleak"
)
// TestMain_ConfigTestReleasesItsGoroutines pins the rule that Main only leaves goroutines running
// when it hands back a Control to stop them with.
//
// A config test gets no Control, so anything it started had nothing to stop it: the lighthouse
// query worker, and a hostname resolver per dns named static host, ran for the life of the
// process. That matters to every embedder that validates a config in process rather than by
// exec'ing, dnclient and the apple clients included, because they do it on each config load and
// the leak accumulates.
func TestMain_ConfigTestReleasesItsGoroutines(t *testing.T) {
defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
l := test.NewLogger()
dir := t.TempDir()
before := time.Now().Add(-time.Hour)
after := time.Now().Add(time.Hour)
ca, _, caKey, caPEM := cert_test.NewTestCaCert(cert.Version2, cert.Curve_CURVE25519, before, after, nil, nil, nil)
networks := []netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")}
_, _, keyPEM, certPEM := cert_test.NewTestCert(
cert.Version2, cert.Curve_CURVE25519, ca, caKey, "config-test", before, after, networks, nil, nil)
caPath := filepath.Join(dir, "ca.pem")
certPath := filepath.Join(dir, "cert.pem")
keyPath := filepath.Join(dir, "key.pem")
require.NoError(t, os.WriteFile(caPath, caPEM, 0o600))
require.NoError(t, os.WriteFile(certPath, certPEM, 0o600))
require.NoError(t, os.WriteFile(keyPath, keyPEM, 0o600))
// A static host by address, not by name: the query worker is the goroutine under test and a
// hostname would drag a real dns lookup into a unit test.
configBody := fmt.Sprintf(`
pki:
ca: %s
cert: %s
key: %s
static_host_map:
"10.0.0.2": ["192.0.2.1:4242"]
lighthouse:
hosts:
- "10.0.0.2"
listen:
host: 127.0.0.1
port: 0
tun:
disabled: true
firewall:
outbound:
- port: any
proto: any
host: any
inbound:
- port: any
proto: any
host: any
`, caPath, certPath, keyPath)
require.NoError(t, os.WriteFile(filepath.Join(dir, "config.yml"), []byte(configBody), 0o600))
c := config.NewC(l)
require.NoError(t, c.Load(dir))
ctrl, err := Main(c, true, "config-test", l, nil)
require.NoError(t, err)
require.Nil(t, ctrl, "a config test hands back nothing to stop, so it must stop itself")
}
+2 -60
View File
@@ -30,9 +30,6 @@ type tun struct {
Routes atomic.Pointer[[]Route] Routes atomic.Pointer[[]Route]
routeTree atomic.Pointer[bart.Table[routing.Gateways]] routeTree atomic.Pointer[bart.Table[routing.Gateways]]
linkAddr *netroute.LinkAddr linkAddr *netroute.LinkAddr
// hostOwned means the fd arrived from the OS, which has already configured addressing, mtu
// and routes for it. NEPacketTunnelProvider on darwin does this.
hostOwned bool
l *slog.Logger l *slog.Logger
} }
@@ -153,48 +150,8 @@ func (t *tun) deviceBytes() (o [16]byte) {
return return
} }
// newTunFromFd adopts a utun the host already created and configured, which is how a darwin func newTunFromFd(_ *config.C, _ *slog.Logger, _ int, _ []netip.Prefix) (*tun, error) {
// network extension is handed its device. Everything about moving packets is shared with newTun, return nil, fmt.Errorf("newTunFromFd not supported in Darwin")
// only the setup differs: the host owns addressing and routing here.
func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
if err := unix.SetNonblock(deviceFd, true); err != nil {
// We own the fd from the moment it is handed to us
_ = unix.Close(deviceFd)
return nil, fmt.Errorf("failed to set the tun fd to non-blocking mode: %w", err)
}
file := os.NewFile(uintptr(deviceFd), "/dev/tun")
t := &tun{
f: file,
Device: utunNameFromFd(deviceFd),
vpnNetworks: vpnNetworks,
DefaultMTU: c.GetInt("tun.mtu", DefaultMTU),
hostOwned: true,
l: l,
}
if err := t.reload(c, true); err != nil {
_ = file.Close()
return nil, err
}
c.RegisterReloadCallback(func(c *config.C) {
if err := t.reload(c, false); err != nil {
util.LogWithContextIfNeeded("failed to reload tun device", err, t.l)
}
})
return t, nil
}
// utunNameFromFd asks the socket what interface it is, for logs. A blank name is not worth
// failing a tunnel over, so an error just leaves it empty.
func utunNameFromFd(fd int) string {
name, err := unix.GetsockoptString(fd, unix.AF_SYS_CONTROL, _UTUN_OPT_IFNAME)
if err != nil {
return ""
}
return name
} }
func (t *tun) Close() error { func (t *tun) Close() error {
@@ -205,12 +162,6 @@ func (t *tun) Close() error {
} }
func (t *tun) Activate() error { func (t *tun) Activate() error {
// The host handed us a configured device. Its addresses, mtu and routes come from the network
// settings it applied, and a sandboxed extension cannot change them anyway.
if t.hostOwned {
return nil
}
devName := t.deviceBytes() devName := t.deviceBytes()
s, err := unix.Socket( s, err := unix.Socket(
@@ -424,11 +375,6 @@ func getLinkAddr(name string) (*netroute.LinkAddr, error) {
} }
func (t *tun) addRoutes(logErrors bool) error { func (t *tun) addRoutes(logErrors bool) error {
// The route tree is still ours, the system routing table is not
if t.hostOwned {
return nil
}
routes := *t.Routes.Load() routes := *t.Routes.Load()
for _, r := range routes { for _, r := range routes {
@@ -458,10 +404,6 @@ func (t *tun) addRoutes(logErrors bool) error {
} }
func (t *tun) removeRoutes(routes []Route) error { func (t *tun) removeRoutes(routes []Route) error {
if t.hostOwned {
return nil
}
for _, r := range routes { for _, r := range routes {
if !r.Install { if !r.Install {
continue continue