mirror of
https://github.com/slackhq/nebula.git
synced 2026-08-16 00:57:01 +02:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 35ac12b0c0 | |||
| c07f28cd04 | |||
| 72bf111209 |
@@ -108,7 +108,12 @@ func (cm *connectionManager) In(h *HostInfo) {
|
|||||||
h.markIn()
|
h.markIn()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Out records outbound traffic and reports whether the local network changed since this tunnel last sent.
|
// OutRelay records relayed traffic, leaving the rebind epoch for the direct path to this host to consume
|
||||||
|
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 {
|
func (cm *connectionManager) Out(h *HostInfo) bool {
|
||||||
return h.markOut(cm.intf.rebindEpoch.Load())
|
return h.markOut(cm.intf.rebindEpoch.Load())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
@@ -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()
|
||||||
|
}
|
||||||
|
|||||||
+14
-14
@@ -238,16 +238,12 @@ 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
|
// The first cache line is everything the packet paths touch.
|
||||||
// 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]
|
||||||
ConnectionState *ConnectionState
|
ConnectionState *ConnectionState
|
||||||
|
|
||||||
// state holds everything the hot paths need to touch per packet, in one word: whether we have seen traffic
|
// Traffic bits, pendingDeletion, and the rebind epoch we last sent under
|
||||||
// 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
|
state atomic.Uint32
|
||||||
|
|
||||||
promoteCounter atomic.Uint32
|
promoteCounter atomic.Uint32
|
||||||
@@ -763,7 +759,7 @@ func (i *HostInfo) TryPromoteBest(preferredRanges []netip.Prefix, ifce *Interfac
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Bits within HostInfo.state. Everything above stateEpochShift is the rebind epoch.
|
// Bits within HostInfo.state, everything above stateEpochShift is the epoch
|
||||||
const (
|
const (
|
||||||
stateIn uint32 = 1 << iota
|
stateIn uint32 = 1 << iota
|
||||||
stateOut
|
stateOut
|
||||||
@@ -773,17 +769,14 @@ const (
|
|||||||
stateEpochShift = 3
|
stateEpochShift = 3
|
||||||
)
|
)
|
||||||
|
|
||||||
// markIn records inbound traffic. Reading first keeps the cache line shared on the common path, where the bit
|
// markIn records inbound traffic
|
||||||
// is already set.
|
|
||||||
func (i *HostInfo) markIn() {
|
func (i *HostInfo) markIn() {
|
||||||
if i.state.Load()&stateIn == 0 {
|
if i.state.Load()&stateIn == 0 {
|
||||||
i.state.Or(stateIn)
|
i.state.Or(stateIn)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// markOut records that we sent on this tunnel under the given rebind epoch. It reports whether the epoch moved
|
// markOut records a send and reports whether the epoch moved, meaning we want a punch from the far side
|
||||||
// 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 {
|
func (i *HostInfo) markOut(epoch uint32) bool {
|
||||||
e := epoch << stateEpochShift
|
e := epoch << stateEpochShift
|
||||||
for {
|
for {
|
||||||
@@ -798,12 +791,19 @@ func (i *HostInfo) markOut(epoch uint32) bool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// sentSinceCheck reports whether anything has been sent since the connection manager last looked.
|
// 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 {
|
func (i *HostInfo) sentSinceCheck() bool {
|
||||||
return i.state.Load()&stateOut != 0
|
return i.state.Load()&stateOut != 0
|
||||||
}
|
}
|
||||||
|
|
||||||
// takeTraffic clears both traffic bits, leaving the epoch alone, and reports what they were.
|
// takeTraffic clears both traffic bits, leaving the epoch alone, and reports what they were
|
||||||
func (i *HostInfo) takeTraffic() (in bool, out bool) {
|
func (i *HostInfo) takeTraffic() (in bool, out bool) {
|
||||||
old := i.state.And(^(stateIn | stateOut))
|
old := i.state.And(^(stateIn | stateOut))
|
||||||
return old&stateIn != 0, old&stateOut != 0
|
return old&stateIn != 0, old&stateOut != 0
|
||||||
|
|||||||
@@ -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))
|
||||||
|
}
|
||||||
|
|||||||
@@ -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,8 +365,7 @@ 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)
|
||||||
// We rebound since this tunnel last sent, so the local network moved. Ask the lighthouse to have the far side
|
// We rebound since this tunnel last sent, ask the lighthouse to get the far side punching at us again
|
||||||
// punch at where we are now, which primes their conntrack the same way a handshake would.
|
|
||||||
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
|
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
|
||||||
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
|
||||||
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
if f.l.Enabled(context.Background(), slog.LevelDebug) {
|
||||||
|
|||||||
+1
-3
@@ -82,9 +82,7 @@ type Interface struct {
|
|||||||
sendRecvErrorConfig recvErrorConfig
|
sendRecvErrorConfig recvErrorConfig
|
||||||
acceptRecvErrorConfig recvErrorConfig
|
acceptRecvErrorConfig recvErrorConfig
|
||||||
|
|
||||||
// rebindEpoch bumps every time the udp listener is rebound, which means the local network moved. Tunnels
|
// Bumped on every udp rebind, tunnels compare it to decide they need a punch from the far side
|
||||||
// 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
|
rebindEpoch atomic.Uint32
|
||||||
version string
|
version string
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user