Compare commits

..

7 Commits

Author SHA1 Message Date
Nate Brown 983784bc41 Group the HostInfo fields the packet paths touch 2026-07-23 16:52:09 -05:00
Nate Brown 30e2735af3 Fold the rebind counter and traffic flags into one atomic word 2026-07-23 16:52:08 -05:00
Nate Brown bce0b4249e Add an e2e Drop exit type and a roaming recovery measurement 2026-07-23 16:52:08 -05:00
Nate Brown 1617897043 v1.11.0 changelog (#1792)
smoke-extra / freebsd-amd64 (push) Failing after 25s
smoke-extra / linux-amd64-ipv6disable (push) Failing after 13s
smoke-extra / netbsd-amd64 (push) Failing after 13s
smoke-extra / openbsd-amd64 (push) Failing after 11s
smoke-extra / linux-386 (push) Failing after 12s
smoke / Run multi node smoke test (push) Failing after 1m39s
Build and test / Static checks (push) Successful in 2m9s
Build and test / Test linux (push) Failing after 1m3s
Build and test / Test linux-boringcrypto (push) Failing after 2m48s
Build and test / Test linux-pkcs11 (push) Failing after 2m0s
Build and test / Cross-build linux-arm (push) Successful in 3m17s
Build and test / Cross-build linux-mips (push) Successful in 4m5s
Build and test / Cross-build linux-other (push) Successful in 3m23s
Build and test / Cross-build windows (push) Successful in 1m4s
Build and test / Cross-build freebsd (push) Successful in 1m42s
Build and test / Cross-build netbsd (push) Successful in 1m38s
Build and test / Cross-build openbsd (push) Successful in 1m43s
Build and test / Cross-build mobile (push) Successful in 3m34s
smoke-extra / Run windows smoke test (push) Has been cancelled
Build and test / Test macos (push) Has been cancelled
Build and test / Test windows (push) Has been cancelled
Build and test / CI status (push) Has been cancelled
2026-07-23 13:13:45 -05:00
Nate Brown f8775bb6ca Use go 1.26 (latest 1.26.5) (#1818) 2026-07-23 10:36:20 -05:00
Nate Brown 15f0f0d5d0 Be less verbose with handshake send errors (#1810) 2026-07-23 09:26:45 -05:00
Nate Brown 7902ce674e Rebind for MacOS (#1816)
Co-authored-by: Jack Doan <me@jackdoan.com>
2026-07-23 09:26:24 -05:00
33 changed files with 1336 additions and 543 deletions
+3 -3
View File
@@ -14,7 +14,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Build
@@ -40,7 +40,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Build
@@ -80,7 +80,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Import certificates
+3 -3
View File
@@ -34,7 +34,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: add hashicorp source
@@ -66,7 +66,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: add hashicorp source
@@ -92,7 +92,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
# WSL2 + Ubuntu so the smoke can run a real linux peer with its own
+1 -1
View File
@@ -22,7 +22,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: build
+4 -4
View File
@@ -22,7 +22,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Install goimports
@@ -42,7 +42,7 @@ jobs:
- name: golangci-lint
uses: golangci/golangci-lint-action@v9
with:
version: v2.5
version: v2.12
test:
name: Test ${{ matrix.name }}
@@ -82,7 +82,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Build
@@ -127,7 +127,7 @@ jobs:
- uses: actions/setup-go@v7
with:
go-version: '1.25'
go-version: '1.26'
check-latest: true
- name: Build ${{ matrix.name }}
+82
View File
@@ -7,6 +7,88 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [1.11.0] - 2026-07-23
See the [v1.11.0](https://github.com/slackhq/nebula/milestone/25?closed=1) milestone for a complete list of changes.
### Breaking
- Logging has switched from logrus to Go's structured `slog`. Log output changes: levels are upper case
(`level=INFO`), trace prints as `level=DEBUG-4`, timestamps are always RFC3339Nano and `logging.timestamp_format`
is ignored, and some messages were reworded. Review any log parsing before upgrading. This is also an API break
for embedders, as constructors now take a `*slog.Logger`. (#1672, #1734, #1621)
- `firewall.inbound_action` and `firewall.outbound_action` (used to set reject vs. drop policy) were each being
applied to the opposite direction, that is now corrected. This only affects how blocked packets are answered, not
which packets the firewall allows or denies. If you set either of these you are getting the behavior of the other
one today and likely want to swap them before upgrading. (#1798)
- On Windows, Nebula now installs WFP PERMIT filters for the nebula adapter and the listener port by default. WFP
sits below Windows Defender Firewall, so any WDF inbound rules you rely on for either will no longer apply. Set
`tun.windows_bypass_wdf` and `listen.windows_bypass_wdf` to false to leave WDF in charge. (#1710)
- On Windows, the nebula device is now set to the `private` network category instead of whatever Windows decided,
which is usually `Public`. This makes the host firewall less restrictive on the overlay. Set
`tun.network_category` to `unset` to keep the old behavior. (#1710)
- Reject packets for non-TCP now use ICMP code 13, communication administratively prohibited, instead of code 3,
port unreachable. Anything keying off the old code needs updating. (#1766, #1768)
- The SSH debug server's profiling commands are now confined to `sshd.sandbox_dir`, which defaults to
`$TMP/nebula-debug`. Relative paths resolve inside it and absolute paths outside it are rejected, so anything
scripting `start-cpu-profile`, `save-heap-profile`, or `save-mutex-profile` with a path elsewhere needs the
directory set. The directory is not created for you. (#1622)
### Added
- Sign the Windows release binaries. (#1718)
- Generate IPv6 reject packets, matching the existing IPv4 behavior. (#1766, #1767, #1768)
- Accept `-` in `nebula-cert` to read from stdin or write to stdout. (#1714)
- Search for both `config.yml` and `config.yaml` in service and command line modes. (#1717)
- Add version labels to the Docker/OCI images. (#1772)
- Rebind the listener and re-query lighthouses on macOS when the underlay network changes, so devices moving
between wifi and wired or between networks recover without waiting for dead tunnel detection. Controlled by
`listen.rebind_on_network_change` (default `true`, not reloadable). (#1816)
### Changed
- Reload the firewall when the unsafe networks in the certificate change. (#1719)
- Reconfigure, start, and stop the stats listener on a config reload instead of requiring a restart. (#1670)
- Update a static host's addresses when they change on reload. (#1713)
- Don't require a port on ICMP firewall rules. (#1609)
- Connection track ICMP traffic. (#1602)
- Return `NODATA` instead of `NXDOMAIN` from the DNS server for a name that exists but has no record of the
requested type, so clients that query `AAAA` first (busybox/Alpine) fall through to `A`. (#1668)
- Record the local host's details in the DNS server. (#1716)
- Install Windows unsafe routes as link routes. (#1709)
- Reduce relay handshake log spam, and only log a handshake send error at error level when the remote list
changes. (#1733, #1765, #1810)
- Start, stop, and reload subsystems (DNS, stats, conntrack, ssh, punchy) cleanly without leaking goroutines. (#1640, #1654, #1661, #1667, #1669, #1708, #1806, #1815)
- `Control` is now safe to stop and wait on from any lifecycle state, and a new `Control.Wait` blocks until nebula
has fully stopped and returns the first fatal reader error. Failed starts release the udp sockets and tun fd
instead of leaking them. (#1794)
- Trigger an immediate lighthouse update when reconnecting to or adding a lighthouse instead of waiting for the next update tick. (#1645)
- Bring the Darwin and OpenBSD tun implementations in line with the other BSDs. (#1703)
- Update to build against go v1.26. (#1818)
- Various dependency updates. (#1586, #1587, #1604, #1617, #1618, #1627, #1628, #1629, #1652, #1664, #1665, #1697, #1721, #1732, #1742, #1743, #1750, #1763, #1771, #1782, #1800, #1807)
### Fixed
- Fix a data race on a host's remote address that could send packets to the wrong address during a roam. (#1773)
- Fix tunnels that could permanently escape connection manager monitoring. (#1752)
- Fix a crash when reloading the SSH server's trusted keys. (#1787)
- Fix hostmap corruption when a host has multiple overlay addresses. Each address now gets its own list instead of
a single shared chain, which also fixes two latent bugs on the add and makePrimary paths. (#1788, #1790)
- Apply `remote_allow_list` IPv4 rules to 4-in-6 mapped addresses. (#1786)
- Don't panic in the DNS server on a short or empty query name. (#1635)
- Advance the replay window on relayed packets so a relay drops replayed frames instead of re-forwarding them. (#1751)
- Fix a race in relay state handling. (#1753)
- Lock replay window updates so concurrent readers can't corrupt it. (#1802)
- Reject malformed handshakes more reliably, including invalid ed25519 key lengths. (#1601, #1756)
- Properly handle `closetunnel` packets. (#1638)
- Fix an IPv6 extension-header length overflow that could make the firewall parse the wrong protocol and ports. (#1789)
- Fix relay re-establishment when a handshake arrives over a relay entry that a one-sided teardown left
`Disestablished`, which silently dropped every send until dead tunnel detection forced a re-handshake. (#1805)
- Don't build new relay state on a tunnel that was just discarded. (#1796)
- Don't delete the wrong pending hostinfo in the handshake manager. (#1811)
- Don't call the packet reader after a UDP error on Darwin. (#1755)
- Open the FreeBSD tun device non blocking. (#1666)
## [1.10.3] - 2026-02-06
### Security
+8 -108
View File
@@ -44,11 +44,6 @@ type connectionManager struct {
inactivityTimeout atomic.Int64
dropInactive atomic.Bool
// Wake-from-sleep handling, sampled once per tick in Start
wakeDetector *wakeDetector
clearOnWake atomic.Bool
wakeClearThreshold atomic.Int64
l *slog.Logger
}
@@ -59,7 +54,6 @@ func newConnectionManagerFromConfig(l *slog.Logger, c *config.C, hm *HostMap, p
punchy: p,
relayUsed: make(map[uint32]struct{}),
relayUsedLock: &sync.RWMutex{},
wakeDetector: newWakeDetector(),
}
cm.reload(c, true)
@@ -104,44 +98,19 @@ func (cm *connectionManager) reload(c *config.C, initial bool) {
)
}
}
if initial || c.HasChanged("tunnels.clear_on_wake") {
old := cm.clearOnWake.Load()
cm.clearOnWake.Store(c.GetBool("tunnels.clear_on_wake", true))
if !initial {
cm.l.Info("Clear on wake setting has changed",
"oldBool", old,
"newBool", cm.clearOnWake.Load(),
)
}
}
if initial || c.HasChanged("tunnels.wake_clear_threshold") {
old := cm.getWakeClearThreshold()
cm.wakeClearThreshold.Store((int64)(c.GetDuration("tunnels.wake_clear_threshold", 30*time.Second)))
if !initial {
cm.l.Info("Wake clear threshold has changed",
"oldDuration", old,
"newDuration", cm.getWakeClearThreshold(),
)
}
}
}
func (cm *connectionManager) getInactivityTimeout() time.Duration {
return (time.Duration)(cm.inactivityTimeout.Load())
}
func (cm *connectionManager) getWakeClearThreshold() time.Duration {
return (time.Duration)(cm.wakeClearThreshold.Load())
}
func (cm *connectionManager) In(h *HostInfo) {
h.in.Store(true)
h.markIn()
}
func (cm *connectionManager) Out(h *HostInfo) {
h.out.Store(true)
// Out records outbound traffic and reports whether the local network changed 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) {
@@ -160,81 +129,13 @@ func (cm *connectionManager) RelayUsed(localIndex uint32) {
// getAndResetTrafficCheck returns if there was any inbound or outbound traffic within the last tick and
// resets the state for this local index
func (cm *connectionManager) getAndResetTrafficCheck(h *HostInfo, now time.Time) (bool, bool) {
in := h.in.Swap(false)
out := h.out.Swap(false)
in, out := h.takeTraffic()
if in || out {
h.lastUsed = now
}
return in, out
}
// checkWake runs once per tick and clears every tunnel when the machine has just returned from system sleep.
// Tunnels rarely survive a suspend: our NAT mappings have expired and our address has usually changed, so every
// established hostinfo is a corpse that will eat 15-20s of traffic checks before the wheel declares it dead.
// Clearing now means the first packet after wake starts a fresh handshake immediately.
//
// The suspend itself costs nothing here: the ticker driving us is frozen with the rest of the process and this
// fires within one tick of resume.
func (cm *connectionManager) checkWake() {
slept, ok := cm.wakeDetector.Sample()
if !ok || slept == 0 {
return
}
// The clock pair is read non-atomically, so scheduling jitter shows up as tiny sub-millisecond "sleeps".
// Keep the floor well above that so a zero/nonsense threshold can't clear tunnels on every tick.
threshold := max(cm.getWakeClearThreshold(), time.Second)
if slept < threshold {
// Short suspends (lid closed and quickly reopened) often come back before NAT state expires; those
// tunnels may well be alive, leave them to the normal traffic checks.
if slept >= time.Second {
cm.l.Debug("Woke from sleep below the clear threshold, leaving tunnels alone",
"sleptFor", slept,
"threshold", threshold,
)
}
return
}
if !cm.clearOnWake.Load() {
cm.l.Info("Woke from sleep, tunnels.clear_on_wake is disabled so tunnels are left to the normal traffic checks", "sleptFor", slept)
return
}
closed := cm.clearAllTunnels()
cm.l.Info("Woke from sleep, cleared tunnels", "sleptFor", slept, "tunnelsCleared", closed)
// Our public address almost certainly changed; get it to the lighthouses as soon as possible so peers can
// find us again. The update rides over a fresh lighthouse handshake. If the network isn't back up yet these
// sends fail harmlessly and the periodic update worker retries within lighthouse.interval.
cm.intf.lightHouse.TriggerUpdate()
}
// clearAllTunnels closes every tunnel in the hostmap locally, without notifying the remotes. It is the wake-from-
// sleep counterpart to Control.CloseAllTunnels: after a suspend the remotes stopped hearing from us long ago, and
// close packets fired into a network that may not even be up yet are wasted, so we only tear down our own state
// and let the next packet to each host start a fresh handshake.
func (cm *connectionManager) clearAllTunnels() int {
cm.hostMap.RLock()
hostinfos := make([]*HostInfo, 0, len(cm.hostMap.Indexes))
for _, h := range cm.hostMap.Indexes {
hostinfos = append(hostinfos, h)
}
cm.hostMap.RUnlock()
for _, h := range hostinfos {
cm.intf.closeTunnel(h)
}
// With every tunnel gone no relay can be in use, drop the usage tracking wholesale.
cm.relayUsedLock.Lock()
clear(cm.relayUsed)
cm.relayUsedLock.Unlock()
return len(hostinfos)
}
func (cm *connectionManager) Start(ctx context.Context) {
clockSource := time.NewTicker(cm.trafficTimer.t.tickDuration)
defer clockSource.Stop()
@@ -249,7 +150,6 @@ func (cm *connectionManager) Start(ctx context.Context) {
return
case now := <-clockSource.C:
cm.checkWake()
cm.trafficTimer.Advance(now)
for {
localIndex, has := cm.trafficTimer.Purge()
@@ -440,7 +340,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim
"tunnelCheck", m{"state": "alive", "method": "passive"},
)
}
hostinfo.pendingDeletion.Store(false)
hostinfo.setPendingDeletion(false)
if mainHostInfo {
decision = tryRehandshake
@@ -463,7 +363,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim
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
hostinfo.logger(cm.l).Info("Tunnel status",
"tunnelCheck", m{"state": "dead", "method": "active"},
@@ -514,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)
return decision, hostinfo, nil
}
+37 -118
View File
@@ -25,6 +25,7 @@ func newTestLighthouse() *LightHouse {
lighthouses := []netip.Addr{}
staticList := map[netip.Addr]struct{}{}
lh.localAddrsFn = func(*LocalAllowList) []netip.Addr { return nil }
lh.lighthouses.Store(&lighthouses)
lh.staticList.Store(&staticList)
@@ -85,25 +86,25 @@ func Test_NewConnectionManagerTest(t *testing.T) {
// We saw traffic out to vpnIp
nc.Out(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.Indexes, hostinfo.localIndexId)
assert.True(t, hostinfo.out.Load())
assert.True(t, hostinfo.in.Load())
assert.True(t, hostinfo.sentSinceCheck())
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
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
// Do another traffic check tick, this host should be pending deletion now
nc.Out(hostinfo)
assert.True(t, hostinfo.out.Load())
assert.True(t, hostinfo.sentSinceCheck())
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
assert.True(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.True(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
@@ -167,33 +168,33 @@ func Test_NewConnectionManagerTest2(t *testing.T) {
// We saw traffic out to vpnIp
nc.Out(hostinfo)
nc.In(hostinfo)
assert.True(t, hostinfo.in.Load())
assert.True(t, hostinfo.out.Load())
assert.False(t, hostinfo.pendingDeletion.Load())
assert.True(t, (hostinfo.state.Load()&stateIn != 0))
assert.True(t, hostinfo.sentSinceCheck())
assert.False(t, hostinfo.isPendingDeletion())
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
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
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
// Do another traffic check tick, this host should be pending deletion now
nc.Out(hostinfo)
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
assert.True(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.True(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
// We saw traffic, should no longer be pending deletion
nc.In(hostinfo)
nc.doTrafficCheck(hostinfo.localIndexId, p, nb, out, time.Now())
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
}
@@ -252,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
nc.Out(hostinfo)
nc.In(hostinfo)
assert.True(t, hostinfo.out.Load())
assert.True(t, hostinfo.in.Load())
assert.True(t, hostinfo.sentSinceCheck())
assert.True(t, (hostinfo.state.Load()&stateIn != 0))
now := time.Now()
decision, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, now)
assert.Equal(t, tryRehandshake, decision)
assert.Equal(t, now, hostinfo.lastUsed)
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*5))
assert.Equal(t, doNothing, decision)
assert.Equal(t, now, hostinfo.lastUsed)
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
// Do another traffic check tick, should still not be pending deletion
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*10))
assert.Equal(t, doNothing, decision)
assert.Equal(t, now, hostinfo.lastUsed)
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
@@ -284,9 +285,9 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) {
decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Minute*10))
assert.Equal(t, closeTunnel, decision)
assert.Equal(t, now, hostinfo.lastUsed)
assert.False(t, hostinfo.pendingDeletion.Load())
assert.False(t, hostinfo.out.Load())
assert.False(t, hostinfo.in.Load())
assert.False(t, hostinfo.isPendingDeletion())
assert.False(t, hostinfo.sentSinceCheck())
assert.False(t, (hostinfo.state.Load()&stateIn != 0))
assert.Contains(t, nc.hostMap.Indexes, hostinfo.localIndexId)
assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0])
}
@@ -501,85 +502,3 @@ func (d *dummyCert) MarshalJSON() ([]byte, error) {
func (d *dummyCert) Copy() cert.Certificate {
return d
}
func TestConnectionManager_WakeClear(t *testing.T) {
l := test.NewLogger()
localrange := netip.MustParsePrefix("10.1.1.1/24")
vpnIp := netip.MustParseAddr("172.1.1.2")
preferredRanges := []netip.Prefix{localrange}
// Very incomplete mock objects
hostMap := newHostMap(l)
hostMap.preferredRanges.Store(&preferredRanges)
cs := &CertState{
initiatingVersion: cert.Version1,
privateKey: []byte{},
v1Cert: &dummyCert{version: cert.Version1},
v1Credential: nil,
}
lh := newTestLighthouse()
ifce := &Interface{
hostMap: hostMap,
inside: &overlaytest.NoopTun{},
outside: &udp.NoopConn{},
firewall: &Firewall{},
lightHouse: lh,
pki: &PKI{},
handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig),
l: l,
}
ifce.pki.cs.Store(cs)
// Create manager
conf := config.NewC(test.NewLogger())
punchy := NewPunchyFromConfig(test.NewLogger(), conf, nil)
nc := newConnectionManagerFromConfig(test.NewLogger(), conf, hostMap, punchy)
nc.intf = ifce
// Drive the wake detector from a fake clock pair
suspended := time.Duration(0)
nc.wakeDetector = &wakeDetector{read: func() (time.Duration, bool) { return suspended, true }}
nc.checkWake() // primes the baseline
addTunnel := func(localIndex uint32) *HostInfo {
hostinfo := &HostInfo{
vpnAddrs: []netip.Addr{vpnIp},
localIndexId: localIndex,
remoteIndexId: 9901,
}
hostinfo.ConnectionState = &ConnectionState{
myCert: &dummyCert{version: cert.Version1},
}
nc.hostMap.unlockedAddHostInfo(hostinfo, ifce)
return hostinfo
}
addTunnel(1099)
nc.RelayUsed(5000)
// No suspend, nothing happens
nc.checkWake()
assert.Contains(t, nc.hostMap.Indexes, uint32(1099))
// A suspend below the threshold leaves tunnels alone
suspended += 5 * time.Second
nc.checkWake()
assert.Contains(t, nc.hostMap.Indexes, uint32(1099))
// A suspend past the threshold clears everything, including relay usage tracking
suspended += time.Hour
nc.checkWake()
assert.Empty(t, nc.hostMap.Indexes)
assert.Empty(t, nc.hostMap.Hosts)
assert.Empty(t, nc.relayUsed)
// With clear_on_wake disabled the tunnels survive a long suspend
addTunnel(1100)
nc.clearOnWake.Store(false)
suspended += time.Hour
nc.checkWake()
assert.Contains(t, nc.hostMap.Indexes, uint32(1100))
assert.Contains(t, nc.hostMap.Hosts, vpnIp)
}
+10 -2
View File
@@ -53,6 +53,7 @@ type Control struct {
statsStart func()
dnsStart func()
lighthouseStart func()
networkChangeStart func(rebind func())
connectionManagerStart func(context.Context)
}
@@ -104,6 +105,9 @@ func (c *Control) Start() error {
if c.dnsStart != nil {
go c.dnsStart()
}
if c.networkChangeStart != nil {
go c.networkChangeStart(c.RebindUDPServer)
}
if c.connectionManagerStart != nil {
go c.connectionManagerStart(c.ctx)
}
@@ -198,13 +202,17 @@ func (c *Control) RebindUDPServer() {
return
}
_ = c.f.outside.Rebind()
// A failure here means we are likely still pinned to the interface we came up on, so the rest of this is
// unlikely to help. Say so instead of silently carrying on as if we rebound.
if err := c.f.outside.Rebind(); err != nil {
c.l.Error("Failed to rebind udp socket", "error", err)
}
// Trigger a lighthouse update, useful for mobile clients that should have an update interval of 0
c.f.lightHouse.SendUpdate()
// 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
+13 -1
View File
@@ -108,7 +108,19 @@ func (c *Control) GetVpnAddrs() []netip.Addr {
}
func (c *Control) GetUDPAddr() netip.AddrPort {
return c.f.outside.(*udp.TesterConn).Addr
return c.f.outside.(*udp.TesterConn).GetAddr()
}
// SetUDPAddr moves this node to a new underlay address, standing in for a laptop waking up on a different
// network. Register the new address with the router as well or nothing will route back.
func (c *Control) SetUDPAddr(addr netip.AddrPort) {
c.f.outside.(*udp.TesterConn).SetAddr(addr)
}
// SetLocalAddrsFn replaces underlay address discovery so a test can advertise its simulated address instead of
// whatever this machine's NICs happen to be. Call it before Start, SendUpdate reads it from the update worker.
func (c *Control) SetLocalAddrsFn(fn func(*LocalAllowList) []netip.Addr) {
c.f.lightHouse.localAddrsFn = fn
}
func (c *Control) KillPendingTunnel(vpnIp netip.Addr) bool {
+225
View File
@@ -0,0 +1,225 @@
//go:build e2e_testing
// +build e2e_testing
package e2e
import (
"net/netip"
"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/header"
"github.com/slackhq/nebula/udp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// reportedAddrs is what the lighthouse would hand a peer asking where vpnAddr is.
func reportedAddrs(t *testing.T, lh *nebula.Control, vpnAddr netip.Addr) []netip.AddrPort {
t.Helper()
cm := lh.QueryLighthouse(vpnAddr)
if cm == nil {
return nil
}
var out []netip.AddrPort
for _, c := range *cm {
out = append(out, c.Reported...)
out = append(out, c.Learned...)
}
return out
}
// waitForLighthouseMsg routes until a lighthouse message lands on lh, or gives up. Reports whether one arrived.
func waitForLighthouseMsg(t *testing.T, r *router.R, lh *nebula.Control, wait time.Duration) bool {
t.Helper()
h := &header.H{}
return r.RouteForAllExitFuncOrTimeout(wait, func(p *udp.Packet, c *nebula.Control) router.ExitType {
if c != lh {
return router.KeepRouting
}
// Punches are a single byte and never parse, they are just not what we are after
if err := h.Parse(p.Data); err != nil {
return router.KeepRouting
}
if h.Type == header.LightHouse {
return router.RouteAndExit
}
return router.KeepRouting
})
}
// A laptop that changes networks has to tell the lighthouse promptly, otherwise the lighthouse keeps handing peers
// the old address and their punches land nowhere. On a long lighthouse interval the only thing that closes that
// window is the rebind, which on darwin the network change monitor drives. The e2e build compiles the monitor out,
// so we call RebindUDPServer directly, which is the same thing the monitor does.
func TestRebindSendsLighthouseUpdate(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{})
lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{
"lighthouse": m{"am_lighthouse": true},
})
// 600s interval, so nothing scheduled can send an update during this test. A rebind is the only thing that can.
myControl, _, _, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", m{
"lighthouse": m{
"hosts": []any{lhVpnIpNet[0].Addr().String()},
"interval": 600,
},
"static_host_map": m{
lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()},
},
})
r := router.NewR(t, lhControl, myControl)
defer r.RenderFlow()
lhControl.Start()
myControl.Start()
// Let the startup registration finish, then clear everything it left behind
require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an initial registration")
r.RouteFor(time.Millisecond * 400)
// Nothing should be talking to the lighthouse on its own now
require.False(t, waitForLighthouseMsg(t, r, lhControl, time.Millisecond*200),
"nothing should reach the lighthouse before the rebind")
myControl.RebindUDPServer()
assert.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5),
"a rebind should push an update to the lighthouse rather than waiting out the interval")
lhControl.Stop()
myControl.Stop()
}
// The other half of a rebind: every live tunnel requeries the lighthouse on its next send. That query is what makes
// the lighthouse tell the peer to punch toward our new address, which is the part that actually revives a tunnel
// whose remote NAT state died while we were on a different network.
func TestRebindRequeriesPeersOnNextSend(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{})
lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{
"lighthouse": m{"am_lighthouse": true},
})
lhCfg := m{
"lighthouse": m{
"hosts": []any{lhVpnIpNet[0].Addr().String()},
"interval": 600,
// Without this the peers advertise this machine's real addresses and then try to punch at them,
// which the router has no route for.
"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", lhCfg)
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "them", "10.128.0.3/24", lhCfg)
r := router.NewR(t, lhControl, myControl, theirControl)
defer r.RenderFlow()
lhControl.Start()
myControl.Start()
theirControl.Start()
r.RouteFor(time.Millisecond * 500)
// Point the peers at each other directly, this test is about the rebind and not about lighthouse discovery
myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("initial")))
r.RouteFor(time.Second)
require.NotNil(t, myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false), "expected a tunnel to them")
r.RouteFor(time.Millisecond * 300)
// Assert on what the peer sees rather than on lighthouse traffic. A query for them makes the lighthouse send
// them a punch notification, which is the whole point. Our own update to the lighthouse sends them nothing,
// so this cannot be satisfied by the update the rebind itself pushes.
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("quiet")))
require.False(t, waitForLighthouseMsg(t, r, theirControl, time.Millisecond*300),
"an ordinary send should not requery the lighthouse")
myControl.RebindUDPServer()
r.RouteFor(time.Millisecond * 300) // let the update the rebind itself sends pass by
myControl.InjectTunPacket(BuildTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("after rebind")))
assert.True(t, waitForLighthouseMsg(t, r, theirControl, time.Second*5),
"the first send after a rebind should requery the lighthouse, which then tells the peer to punch at us")
lhControl.Stop()
myControl.Stop()
theirControl.Stop()
}
// The scenario this whole thing exists for: a laptop sleeps at the office and wakes up at home on a new address.
// Until it tells the lighthouse, the lighthouse keeps handing peers the office address, so their punches land
// nowhere and the tunnel stays dead. On a long interval the rebind is the only thing that closes that window.
func TestRebindAdvertisesNewAddressAfterMove(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{})
lhControl, lhVpnIpNet, lhUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "lh", "10.128.0.1/24", m{
"lighthouse": m{"am_lighthouse": true},
})
myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version2, ca, caKey, "me", "10.128.0.2/24", m{
"lighthouse": m{
"hosts": []any{lhVpnIpNet[0].Addr().String()},
"interval": 600,
},
"static_host_map": m{
lhVpnIpNet[0].Addr().String(): []any{lhUdpAddr.String()},
},
})
// Advertise wherever we currently are rather than this machine's real NICs, read fresh each time so a move
// is picked up.
myControl.SetLocalAddrsFn(func(*nebula.LocalAllowList) []netip.Addr {
return []netip.Addr{myControl.GetUDPAddr().Addr()}
})
r := router.NewR(t, lhControl, myControl)
defer r.RenderFlow()
lhControl.Start()
myControl.Start()
require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an initial registration")
r.RouteFor(time.Millisecond * 400)
require.Contains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), myUdpAddr,
"the lighthouse should know the address we started on")
// Wake up somewhere else
newAddr := netip.MustParseAddrPort("10.0.0.99:4242")
myControl.SetUDPAddr(newAddr)
r.AddRoute(newAddr.Addr(), newAddr.Port(), myControl)
// Nothing has told the lighthouse, and with interval 600 nothing scheduled will
r.RouteFor(time.Millisecond * 400)
require.NotContains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), newAddr,
"the lighthouse should still be handing out the old address before the rebind")
myControl.RebindUDPServer()
require.True(t, waitForLighthouseMsg(t, r, lhControl, time.Second*5), "expected an update after the rebind")
r.RouteFor(time.Millisecond * 400)
assert.Contains(t, reportedAddrs(t, lhControl, myVpnIpNet[0].Addr()), newAddr,
"after the rebind the lighthouse should hand peers our new address")
lhControl.Stop()
myControl.Stop()
}
+136
View File
@@ -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
}
+148 -29
View File
@@ -114,6 +114,28 @@ type packet struct {
packet *udp.Packet
tun bool // a packet pulled off a tun device
rx bool // the packet was received by a udp device
// h is the nebula header, parsed once when the packet is recorded. parseErr says why there isn't one, which
// the flow log reports rather than hiding. Punchy sends a single byte, so an unparseable packet is normal.
h header.H
parseErr error
}
// fromAddr and toAddr are the addresses this packet actually travelled between. Reading them off the control
// instead would misreport the whole history once a test moves a node. Tun packets are synthesized without
// addresses, so they fall back to the control.
func (p *packet) fromAddr() netip.AddrPort {
if p.tun || !p.packet.From.IsValid() {
return p.from.GetUDPAddr()
}
return p.packet.From
}
func (p *packet) toAddr() netip.AddrPort {
if p.tun || !p.packet.To.IsValid() {
return p.to.GetUDPAddr()
}
return p.packet.To
}
func (p *packet) WasReceived() {
@@ -131,6 +153,9 @@ const (
ExitNow ExitType = 1
// RouteAndExit routes this packet and exits immediately afterwards
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
@@ -141,7 +166,9 @@ type ExitFunc func(packet *udp.Packet, receiver *nebula.Control) ExitType
func NewR(t testing.TB, controls ...*nebula.Control) *R {
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)
}
@@ -152,7 +179,7 @@ func NewR(t testing.TB, controls ...*nebula.Control) *R {
outNat: make(map[outNatKey]netip.AddrPort),
flow: []flowEntry{},
ignoreFlows: []ignoreFlow{},
fn: filepath.Join("mermaid", fmt.Sprintf("%s.md", t.Name())),
fn: fn,
t: t,
cancelRender: cancel,
}
@@ -249,7 +276,7 @@ func (r *R) renderFlow() {
continue
}
addr := e.packet.from.GetUDPAddr()
addr := e.packet.fromAddr()
if _, ok := participants[addr]; ok {
continue
}
@@ -268,7 +295,6 @@ func (r *R) renderFlow() {
}
// Print packets
h := &header.H{}
for _, e := range r.flow {
if e.packet == nil {
//fmt.Fprintf(f, " note over %s: %s\n", strings.Join(participantsVals, ", "), e.note)
@@ -280,21 +306,22 @@ func (r *R) renderFlow() {
fmt.Fprintln(f, r.formatUdpPacket(p))
} else {
if err := h.Parse(p.packet.Data); err != nil {
panic(err)
}
line := "--x"
if p.rx {
line = "->>"
}
fmt.Fprintf(f,
" %s%s%s: %s(%s), index %v, counter: %v\n",
normalizeName(p.from.GetUDPAddr().String()),
detail := fmt.Sprintf("%s(%s), index %v, counter: %v",
p.h.TypeName(), p.h.SubTypeName(), p.h.RemoteIndex, p.h.MessageCounter)
if p.parseErr != nil {
detail = fmt.Sprintf("unparsed, %v (%d bytes)", p.parseErr, len(p.packet.Data))
}
fmt.Fprintf(f, " %s%s%s: %s\n",
normalizeName(p.fromAddr().String()),
line,
normalizeName(p.to.GetUDPAddr().String()),
h.TypeName(), h.SubTypeName(), h.RemoteIndex, h.MessageCounter,
normalizeName(p.toAddr().String()),
detail,
)
}
}
@@ -408,29 +435,34 @@ func (r *R) unlockedInjectFlow(from, to *nebula.Control, p *udp.Packet, tun bool
r.renderHostmaps(fmt.Sprintf("Packet %v", len(r.flow)))
if len(r.ignoreFlows) > 0 {
var h header.H
err := h.Parse(p.Data)
if err != nil {
panic(err)
}
var h header.H
var parseErr error
if !tun {
parseErr = h.Parse(p.Data)
}
for _, i := range r.ignoreFlows {
if !tun {
if i.messageType == h.Type && i.subType == h.Subtype {
return nil
}
} else if i.tun.HasValue && i.tun.IsTrue {
// Decide before copying, the copy comes from a freelist and an ignored packet would never be released
for _, i := range r.ignoreFlows {
if tun {
if i.tun.HasValue && i.tun.IsTrue {
return nil
}
continue
}
// A packet we could not parse has no type to match against, so no rule can ignore it
if parseErr == nil && i.messageType == h.Type && i.subType == h.Subtype {
return nil
}
}
fp := &packet{
from: from,
to: to,
packet: p.Copy(),
tun: tun,
from: from,
to: to,
packet: p.Copy(),
tun: tun,
h: h,
parseErr: parseErr,
}
r.flow = append(r.flow, flowEntry{packet: fp})
@@ -660,6 +692,10 @@ func (r *R) RouteExitFunc(sender *nebula.Control, whatDo ExitFunc) {
p.Release()
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:
fp := r.unlockedInjectFlow(sender, receiver, p, false)
receiver.InjectUDPPacket(p)
@@ -690,6 +726,85 @@ func (r *R) RouteUntilAfterMsgType(sender *nebula.Control, msgType header.Messag
})
}
// RouteFor routes everything that shows up for the given duration and then returns. Use it to let a test settle
// deterministically rather than sleeping and hoping: a single FlushAll races a completing handshake, which queues
// more packets right behind it.
func (r *R) RouteFor(d time.Duration) {
r.RouteForAllExitFuncOrTimeout(d, func(*udp.Packet, *nebula.Control) ExitType {
return KeepRouting
})
}
// RouteForAllExitFuncOrTimeout is RouteForAllExitFunc with a deadline, reporting whether whatDo asked to exit
// before time ran out. The unbounded version blocks forever on a quiet network, so this is what a test needs to
// assert that something does NOT happen, or to route for a fixed settling period.
func (r *R) RouteForAllExitFuncOrTimeout(timeout time.Duration, whatDo ExitFunc) bool {
sc := make([]reflect.SelectCase, 0, len(r.controls)+1)
cm := make([]*nebula.Control, 0, len(r.controls))
for _, c := range r.controls {
sc = append(sc, reflect.SelectCase{
Dir: reflect.SelectRecv,
Chan: reflect.ValueOf(c.GetUDPTxChan()),
Send: reflect.Value{},
})
cm = append(cm, c)
}
timer := time.NewTimer(timeout)
defer timer.Stop()
sc = append(sc, reflect.SelectCase{
Dir: reflect.SelectRecv,
Chan: reflect.ValueOf(timer.C),
Send: reflect.Value{},
})
for {
x, rx, _ := reflect.Select(sc)
if x == len(cm) {
return false
}
r.Lock()
p := rx.Interface().(*udp.Packet)
receiver := r.getControl(cm[x].GetUDPAddr(), p.To, p)
if receiver == nil {
r.Unlock()
panic("Can't RouteForAllExitFuncOrTimeout for host: " + p.To.String())
}
e := whatDo(p, receiver)
switch e {
case ExitNow:
r.Unlock()
p.Release()
return true
case RouteAndExit:
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
receiver.InjectUDPPacket(p)
fp.WasReceived()
r.Unlock()
p.Release()
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:
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
receiver.InjectUDPPacket(p)
fp.WasReceived()
default:
panic(fmt.Sprintf("Unknown exitFunc return: %v", e))
}
r.Unlock()
p.Release()
}
}
func (r *R) RouteForAllUntilAfterMsgTypeTo(receiver *nebula.Control, msgType header.MessageType, subType header.MessageSubType) {
h := &header.H{}
r.RouteForAllExitFunc(func(p *udp.Packet, r *nebula.Control) ExitType {
@@ -782,6 +897,10 @@ func (r *R) RouteForAllExitFunc(whatDo ExitFunc) {
p.Release()
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:
fp := r.unlockedInjectFlow(cm[x], receiver, p, false)
receiver.InjectUDPPacket(p)
+8 -14
View File
@@ -146,6 +146,14 @@ listen:
# Default true; set to false to leave WDF in charge of inbound decisions on the listener port. Not reloadable.
#windows_bypass_wdf: true
# On macOS only
# macOS scopes the udp socket to the interface it was created on, so moving between networks (wifi to wired,
# office to home) leaves Nebula sending out an interface that no longer has a route. When true, Nebula watches
# the routing socket and rebinds the listener once the change settles.
# iOS does not use this, the host app drives the same rebind itself.
# Default true. Not reloadable.
#rebind_on_network_change: true
# By default, Nebula replies to packets it has no tunnel for with a "recv_error" packet. This packet helps speed up reconnection
# in the case that Nebula on either side did not shut down cleanly. This response can be abused as a way to discover if Nebula is running
# on a host though. This option lets you configure if you want to send "recv_error" packets always, never, or only to private network remotes.
@@ -390,20 +398,6 @@ logging:
# This setting is reloadable
#inactivity_timeout: 10m
# clear_on_wake controls whether all tunnels are immediately torn down (locally, without notifying the remotes)
# when the machine detects it has just woken from system sleep. Tunnels rarely survive a suspend: NAT mappings
# expire and the machine's address usually changes, so waiting for the normal liveness checks costs 15-20 seconds
# of black-holed traffic per tunnel after wake. Clearing them means the first packet after wake starts a fresh
# handshake right away.
# This setting is reloadable
#clear_on_wake: true
# wake_clear_threshold is the minimum time the machine must have been suspended for clear_on_wake to act.
# Suspends shorter than this often come back before NAT state expires, so those tunnels may still be alive and
# are left to the normal liveness checks. Values below 1s are treated as 1s.
# This setting is reloadable
#wake_clear_threshold: 30s
# Nebula security group configuration
firewall:
# Action to take when a packet is not allowed by the firewall rules.
+1 -1
View File
@@ -1,6 +1,6 @@
module github.com/slackhq/nebula
go 1.25.0
go 1.26.0
require (
dario.cat/mergo v1.0.2
+7 -1
View File
@@ -295,7 +295,13 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
hm.messageMetrics.Tx(header.Handshake, hh.machine.Subtype(), 1)
err := hm.outside.WriteTo(stage0, addr)
if err != nil {
hostinfo.logger(hm.l).Error("Failed to send handshake message",
// These repeat every attempt, so match the success log below and only shout when the remotes changed
level := slog.LevelDebug
if remotesHaveChanged {
level = slog.LevelError
}
hostinfo.logger(hm.l).Log(context.Background(), level, "Failed to send handshake message",
"udpAddr", addr,
"initiatorIndex", hostinfo.localIndexId,
"handshake", hsFields,
+75 -13
View File
@@ -238,18 +238,30 @@ const (
)
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]
remotes *RemoteList
promoteCounter atomic.Uint32
ConnectionState *ConnectionState
remoteIndexId uint32
localIndexId uint32
// 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
localIndexId uint32
remotes *RemoteList
// 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
// vpn networks but were removed because they are not usable
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 *bart.Table[NetworkType]
relayState RelayState
@@ -262,11 +274,6 @@ type HostInfo struct {
// This is used to limit lighthouse re-queries in chatty clients
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
// 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
@@ -275,9 +282,6 @@ type HostInfo struct {
lastRoam time.Time
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.
// This value will be behind against actual tunnel utilization in the hot path.
// 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.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
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 {
if i.ConnectionState != nil {
return i.ConnectionState.peerCert
+5 -10
View File
@@ -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)
out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c)
f.connectionManager.Out(hostinfo)
// 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.
// 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.
if f.connectionManager.Out(hostinfo) && t != header.CloseTunnel {
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
hostinfo.lastRebindCount = f.rebindCount
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,
)
}
@@ -408,7 +403,7 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType
if err != nil {
hostinfo.logger(f.l).Error("Failed to write outgoing packet",
"error", err,
"udpAddr", remote,
"udpAddr", hr,
)
}
} else {
+4 -2
View File
@@ -82,8 +82,10 @@ type Interface struct {
sendRecvErrorConfig recvErrorConfig
acceptRecvErrorConfig recvErrorConfig
// rebindCount is used to decide if an active tunnel should trigger a punch notification through a lighthouse
rebindCount int8
// rebindEpoch bumps every time the udp listener is rebound, which means the local network moved. Tunnels
// 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
conntrackCacheTimeout time.Duration
+9 -1
View File
@@ -36,6 +36,10 @@ type LightHouse struct {
myVpnNetworksTable *bart.Lite
punchy *Punchy
// localAddrsFn enumerates the underlay addresses we advertise. It is a field so tests can supply simulated
// addresses rather than whatever this machine's NICs happen to be. Set it before Start.
localAddrsFn func(*LocalAllowList) []netip.Addr
// Local cache of answers from light houses
// map of vpn addr to answers
addrMap map[netip.Addr]*RemoteList
@@ -107,6 +111,10 @@ func NewLightHouseFromConfig(ctx context.Context, l *slog.Logger, c *config.C, c
queryChan: make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 64)),
l: l,
}
h.localAddrsFn = func(al *LocalAllowList) []netip.Addr {
return localAddrs(h.l, al)
}
lighthouses := make([]netip.Addr, 0)
h.lighthouses.Store(&lighthouses)
staticList := make(map[netip.Addr]struct{})
@@ -918,7 +926,7 @@ func (lh *LightHouse) SendUpdate() {
}
lal := lh.GetLocalAllowList()
for _, e := range localAddrs(lh.l, lal) {
for _, e := range lh.localAddrsFn(lal) {
if lh.myVpnNetworksTable.Contains(e) {
continue
}
+3
View File
@@ -268,6 +268,8 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
attachCommands(l, c, ssh, ifce)
networkChanges := udp.NewNetworkChangeMonitor(ctx, l, c)
return &Control{
state: StateReady,
f: ifce,
@@ -278,6 +280,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
statsStart: stats.Start,
dnsStart: ds.Start,
lighthouseStart: lightHouse.StartUpdateWorker,
networkChangeStart: networkChanges.Start,
connectionManagerStart: connManager.Start,
}, nil
}
+61
View File
@@ -0,0 +1,61 @@
package udp
import (
"context"
"log/slog"
"github.com/slackhq/nebula/config"
)
// NetworkChangeMonitor rebinds the udp listener when the local network moves out from under it.
//
// Detection lives here in the udp package, next to the socket it concerns and the platform matrix that already knows
// which sockets go stale. What to do about a change — updating the lighthouse, requerying tunnels — is not the udp
// package's business, so Start takes the reaction as a plain function. Passing it at Start rather than holding it
// keeps this package from referencing whatever owns the rebind.
//
// On platforms whose sockets do not go stale, watchNetworkChanges hands back a nil channel and Start returns.
type NetworkChangeMonitor struct {
l *slog.Logger
ctx context.Context
enabled bool
}
// NewNetworkChangeMonitor builds a monitor for local network changes. The returned monitor is always usable: Start
// is safe to call unconditionally, it no-ops when disabled or on a platform that does not need it.
func NewNetworkChangeMonitor(ctx context.Context, l *slog.Logger, c *config.C) *NetworkChangeMonitor {
return &NetworkChangeMonitor{
l: l,
ctx: ctx,
enabled: c.GetBool("listen.rebind_on_network_change", true),
}
}
// Start watches for network changes until the context is cancelled, calling rebind once per settled change. It
// blocks, so callers run it in a goroutine, and it no-ops when disabled, unsupported, or with nothing to rebind.
func (m *NetworkChangeMonitor) Start(rebind func()) {
if !m.enabled || rebind == nil || m.ctx.Err() != nil {
return
}
changes, err := watchNetworkChanges(m.ctx, m.l)
if err != nil {
// Not fatal. Everything else still works, we just won't notice a network change on our own.
m.l.Error("Failed to watch for network changes, will not rebind the udp listener when the network moves",
"error", err,
)
return
}
if changes == nil {
// This platform's sockets don't go stale, so there is nothing to watch for.
return
}
m.l.Info("Watching for network changes to rebind the udp listener")
for range changes {
m.l.Info("Local network changed, rebinding the udp listener")
rebind()
}
}
+164
View File
@@ -0,0 +1,164 @@
//go:build darwin && !ios && !e2e_testing
// +build darwin,!ios,!e2e_testing
package udp
import (
"context"
"encoding/binary"
"errors"
"log/slog"
"os"
"time"
"golang.org/x/sys/unix"
)
const (
// netChangeSettleWindow is how long we keep swallowing routing messages after the first interesting one. A
// single network change is never a single message, it is a burst: the link drops, addresses go away, new ones
// arrive, routes get rewritten. Reporting part way through that just means reporting again.
netChangeSettleWindow = time.Second
// netChangeReadBuffer is sized well past any rt_msghdr plus its addresses. A short read would be discarded by
// the kernel, so being generous here is how we avoid missing a message.
netChangeReadBuffer = 4096
)
// watchNetworkChanges reports when the local network moves out from under us, so the listener can be rebound.
//
// Darwin scopes a udp socket to whatever interface it came up on. Move between networks and we keep sending out an
// interface that no longer has a route, which surfaces as an instant "no route to host" with no packet ever leaving
// the box. Rebind clears that, but only if something notices the change and calls it. iOS has always been told by
// the host app off NWPathMonitor. This is the equivalent for everything else that runs on darwin.
//
// The returned channel is buffered and coalescing: a send is dropped if one is already pending, since both mean the
// same thing to a reader. It is closed when ctx is cancelled or the routing socket fails, so a caller can simply
// range over it. Platforms whose sockets do not need rebinding return a nil channel and no error.
func watchNetworkChanges(ctx context.Context, l *slog.Logger) (<-chan struct{}, error) {
sock, err := openRouteSocket()
if err != nil {
return nil, err
}
changes := make(chan struct{}, 1)
go func() {
defer close(changes)
defer func() { _ = sock.Close() }()
// Closing the socket is what unblocks the read in watchRouteSocket, so this turns cancellation into a
// close. It is scoped to this call so it cannot outlive the watch it belongs to.
done := make(chan struct{})
defer close(done)
go func() {
select {
case <-ctx.Done():
_ = sock.Close()
case <-done:
}
}()
watchRouteSocket(l, sock, changes)
}()
return changes, nil
}
// watchRouteSocket blocks reading the routing socket, reporting once per settled burst of changes. It returns when
// the socket is closed, which is how cancellation gets us out of here.
func watchRouteSocket(l *slog.Logger, sock *os.File, changes chan<- struct{}) {
buf := make([]byte, netChangeReadBuffer)
for {
n, err := sock.Read(buf)
if err != nil {
logRouteSocketError(l, err)
return
}
if !isNetworkChange(buf[:n]) {
continue
}
// Swallow the rest of the burst. The deadline is absolute and not extended by what arrives, so this always
// ends after the settle window no matter how chatty the socket is. Changes that land after the window
// simply produce another report, which is the correct outcome anyway.
deadline := time.Now().Add(netChangeSettleWindow)
for {
if err = sock.SetReadDeadline(deadline); err != nil {
logRouteSocketError(l, err)
return
}
if _, err = sock.Read(buf); err != nil {
if os.IsTimeout(err) {
break
}
logRouteSocketError(l, err)
return
}
}
if err = sock.SetReadDeadline(time.Time{}); err != nil {
logRouteSocketError(l, err)
return
}
select {
case changes <- struct{}{}:
default:
// One already pending, and a second "the network moved" tells the reader nothing new.
}
}
}
// logRouteSocketError reports a routing socket failure unless it is just us shutting the socket down.
func logRouteSocketError(l *slog.Logger, err error) {
if errors.Is(err, os.ErrClosed) {
return
}
l.Error("Error reading the routing socket, will no longer notice local network changes", "error", err)
}
// openRouteSocket returns the routing socket as a non blocking os.File. Going through os.File puts reads on the go
// poller, which buys us both a working read deadline and a Close that unblocks a read in progress.
func openRouteSocket() (*os.File, error) {
fd, err := unix.Socket(unix.AF_ROUTE, unix.SOCK_RAW, unix.AF_UNSPEC)
if err != nil {
return nil, err
}
if err = unix.SetNonblock(fd, true); err != nil {
_ = unix.Close(fd)
return nil, err
}
return os.NewFile(uintptr(fd), "route"), nil
}
// isNetworkChange reports whether a routing message means our local addressing may have moved out from under us.
//
// We read the header instead of parsing the message because the type is the only part we need, and a full parse can
// fail on shapes we don't care about, which would turn "a message I can't parse" into "a change I missed".
// rt_msghdr, if_msghdr and ifa_msghdr all begin with the same three fields, so this is the same for every type.
func isNetworkChange(msg []byte) bool {
if len(msg) < 4 {
return false
}
// u_short msglen, u_char version, u_char type
if int(binary.NativeEndian.Uint16(msg[0:2])) > len(msg) || msg[2] != unix.RTM_VERSION {
return false
}
switch msg[3] {
case unix.RTM_NEWADDR, unix.RTM_DELADDR, unix.RTM_IFINFO:
// An address arrived or left, or a link changed state. Anything else on this socket is either a route
// churning underneath us, which a rebind doesn't help with, or unrelated traffic.
return true
default:
return false
}
}
+244
View File
@@ -0,0 +1,244 @@
//go:build darwin && !ios && !e2e_testing
// +build darwin,!ios,!e2e_testing
package udp
import (
"context"
"encoding/binary"
"os"
"testing"
"time"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/goleak"
"golang.org/x/sys/unix"
)
// routeMsg builds the first four bytes of a routing message, which is all isNetworkChange reads.
func routeMsg(msgType uint8, extra int) []byte {
msg := make([]byte, 4+extra)
binary.NativeEndian.PutUint16(msg[0:2], uint16(len(msg)))
msg[2] = unix.RTM_VERSION
msg[3] = msgType
return msg
}
func TestIsNetworkChange(t *testing.T) {
// The three that mean our addressing may have moved
assert.True(t, isNetworkChange(routeMsg(unix.RTM_NEWADDR, 0)))
assert.True(t, isNetworkChange(routeMsg(unix.RTM_DELADDR, 0)))
assert.True(t, isNetworkChange(routeMsg(unix.RTM_IFINFO, 0)))
// Route churn is not something a rebind helps with
assert.False(t, isNetworkChange(routeMsg(unix.RTM_ADD, 0)))
assert.False(t, isNetworkChange(routeMsg(unix.RTM_DELETE, 0)))
assert.False(t, isNetworkChange(routeMsg(unix.RTM_GET, 0)))
// Garbage must not be mistaken for a change
assert.False(t, isNetworkChange(nil), "empty")
assert.False(t, isNetworkChange([]byte{0, 0, 0}), "short header")
wrongVersion := routeMsg(unix.RTM_NEWADDR, 0)
wrongVersion[2] = unix.RTM_VERSION + 1
assert.False(t, isNetworkChange(wrongVersion), "wrong rtm_version")
lying := routeMsg(unix.RTM_NEWADDR, 0)
binary.NativeEndian.PutUint16(lying[0:2], 512)
assert.False(t, isNetworkChange(lying), "msglen longer than what we read")
}
// socketPair returns a connected pair of datagram sockets, the first wrapped the same way the routing socket is. It
// stands in for the kernel so the watch loop can be driven with synthetic messages.
func socketPair(t *testing.T) (*os.File, int) {
t.Helper()
fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0)
require.NoError(t, err)
require.NoError(t, unix.SetNonblock(fds[0], true))
f := os.NewFile(uintptr(fds[0]), "route")
t.Cleanup(func() {
_ = f.Close()
_ = unix.Close(fds[1])
})
return f, fds[1]
}
func TestWatchRouteSocketCoalescesABurst(t *testing.T) {
sock, kernel := socketPair(t)
changes := make(chan struct{}, 1)
done := make(chan struct{})
go func() {
watchRouteSocket(test.NewLogger(), sock, changes)
close(done)
}()
// One network change is a burst of messages. All of these land inside the settle window, so they must produce
// exactly one report rather than one apiece.
for range 5 {
_, err := unix.Write(kernel, routeMsg(unix.RTM_NEWADDR, 8))
require.NoError(t, err)
}
// Uninteresting messages in the middle of a burst must not add a report of their own either.
_, err := unix.Write(kernel, routeMsg(unix.RTM_ADD, 8))
require.NoError(t, err)
select {
case <-changes:
case <-time.After(netChangeSettleWindow * 4):
t.Fatal("a burst should have reported a change")
}
// Nothing more from that burst
select {
case <-changes:
t.Fatal("a burst should report exactly once")
case <-time.After(netChangeSettleWindow):
}
// A change after the window has closed is a separate event and gets its own report.
_, err = unix.Write(kernel, routeMsg(unix.RTM_IFINFO, 8))
require.NoError(t, err)
select {
case <-changes:
case <-time.After(netChangeSettleWindow * 4):
t.Fatal("a later change should report again")
}
// Closing the socket is how the real thing shuts down
require.NoError(t, sock.Close())
select {
case <-done:
case <-time.After(time.Second * 5):
t.Fatal("watchRouteSocket did not return after the socket was closed")
}
}
func TestWatchRouteSocketIgnoresUninterestingMessages(t *testing.T) {
sock, kernel := socketPair(t)
changes := make(chan struct{}, 1)
done := make(chan struct{})
go func() {
watchRouteSocket(test.NewLogger(), sock, changes)
close(done)
}()
for _, msgType := range []uint8{unix.RTM_ADD, unix.RTM_DELETE, unix.RTM_GET, unix.RTM_MISS} {
_, err := unix.Write(kernel, routeMsg(msgType, 8))
require.NoError(t, err)
}
select {
case <-changes:
t.Fatal("route churn alone must not report a change")
case <-time.After(netChangeSettleWindow * 2):
}
require.NoError(t, sock.Close())
select {
case <-done:
case <-time.After(time.Second * 5):
t.Fatal("watchRouteSocket did not return after the socket was closed")
}
}
// TestWatchRouteSocketDropsRatherThanBlocks covers the coalescing send. A reader that is busy rebinding must not
// wedge the watcher, and a second pending "the network moved" tells it nothing new anyway.
func TestWatchRouteSocketDropsRatherThanBlocks(t *testing.T) {
sock, kernel := socketPair(t)
changes := make(chan struct{}, 1)
done := make(chan struct{})
go func() {
watchRouteSocket(test.NewLogger(), sock, changes)
close(done)
}()
// Nobody is reading changes, so after the first report the buffer is full for the rest of this test
for range 3 {
_, err := unix.Write(kernel, routeMsg(unix.RTM_NEWADDR, 8))
require.NoError(t, err)
time.Sleep(netChangeSettleWindow + time.Millisecond*250)
}
// The watcher must still be alive and responsive to a close
require.NoError(t, sock.Close())
select {
case <-done:
case <-time.After(time.Second * 5):
t.Fatal("watchRouteSocket wedged on a full channel")
}
assert.Len(t, changes, 1, "the pending report should have coalesced, not queued")
}
// TestWatchNetworkChangesStopsWithContext covers the detection path against a real routing socket, including that
// cancelling the context closes the channel so a ranging caller falls out of its loop.
func TestWatchNetworkChangesStopsWithContext(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
changes, err := watchNetworkChanges(ctx, test.NewLogger())
require.NoError(t, err)
require.NotNil(t, changes, "darwin should support watching")
drained := make(chan struct{})
go func() {
for range changes {
}
close(drained)
}()
cancel()
select {
case <-drained:
case <-time.After(time.Second * 5):
t.Fatal("cancelling the context should close the changes channel")
}
}
// TestNetworkChangeMonitorStopsWithContext drives the whole monitor against a real routing socket: Start must block
// watching, and cancelling the context (which is all Control does on shutdown, it never stops the monitor directly)
// must return it and clean up the watch goroutines.
func TestNetworkChangeMonitorStopsWithContext(t *testing.T) {
// IgnoreCurrent because other tests in this package leave readers running; we only care about what this test
// leaks itself.
defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
ctx, cancel := context.WithCancel(context.Background())
l := test.NewLogger()
c := config.NewC(l)
require.NoError(t, c.LoadString("listen:\n rebind_on_network_change: true\n"))
m := NewNetworkChangeMonitor(ctx, l, c)
done := make(chan struct{})
go func() {
m.Start(func() {})
close(done)
}()
// Start should be sitting on the routing socket, not have fallen out. If it returned early it either failed to
// watch or no-op'd, both of which we want to catch.
select {
case <-done:
t.Fatal("Start returned instead of watching")
case <-time.After(time.Millisecond * 250):
}
cancel()
select {
case <-done:
case <-time.After(time.Second * 5):
t.Fatal("Start did not return after the context was cancelled")
}
// Starting again after the context is dead must not open anything.
m.Start(func() {})
}
+22
View File
@@ -0,0 +1,22 @@
//go:build !darwin || ios || e2e_testing
// +build !darwin ios e2e_testing
package udp
import (
"context"
"log/slog"
)
// watchNetworkChanges is a no-op outside of darwin.
//
// Darwin is the platform that scopes a udp socket to the interface it came up on, so it is the platform whose socket
// goes stale when the local network changes. Everywhere else Rebind has nothing to do, so there is nothing to watch
// for. iOS is excluded on purpose even though it is darwin: the host app already drives the rebind off NWPathMonitor,
// and two things racing to rebind the same socket is worse than one.
//
// A nil channel means "not supported here", which callers must treat as "do not start a watcher" rather than
// selecting on it, since a receive from a nil channel blocks forever.
func watchNetworkChanges(_ context.Context, _ *slog.Logger) (<-chan struct{}, error) {
return nil, nil
}
+39
View File
@@ -0,0 +1,39 @@
package udp
import (
"context"
"testing"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func newMonitor(t *testing.T, ctx context.Context, cfg string) *NetworkChangeMonitor {
t.Helper()
l := test.NewLogger()
c := config.NewC(l)
require.NoError(t, c.LoadString(cfg))
return NewNetworkChangeMonitor(ctx, l, c)
}
func TestNetworkChangeMonitorDefaultsOn(t *testing.T) {
// Says nothing about rebinding, so this covers the default.
m := newMonitor(t, context.Background(), "listen:\n host: 0.0.0.0\n")
assert.True(t, m.enabled, "should default to on")
}
func TestNetworkChangeMonitorDisabledIsANoOp(t *testing.T) {
m := newMonitor(t, context.Background(), "listen:\n rebind_on_network_change: false\n")
require.False(t, m.enabled)
// Must return without opening a socket. If it watched anything this would block.
m.Start(func() {})
}
func TestNetworkChangeMonitorNilRebindIsANoOp(t *testing.T) {
// Nothing to rebind, so there is no point watching, on any platform.
m := newMonitor(t, context.Background(), "listen:\n rebind_on_network_change: true\n")
m.Start(nil)
}
+4 -5
View File
@@ -187,6 +187,9 @@ func (u *StdConn) SupportsMultipleReaders() bool {
return false
}
// Rebind clears the interface the kernel scoped this socket to, so that sends are routed against the current
// routing table instead of the interface we happened to be on when the socket was created. Darwin pins sockets
// this way on its own, which is what strands us after the underlying network changes.
func (u *StdConn) Rebind() error {
var err error
if u.isV4 {
@@ -195,9 +198,5 @@ func (u *StdConn) Rebind() error {
err = syscall.SetsockoptInt(int(u.sysFd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, 0)
}
if err != nil {
u.l.Error("Failed to rebind udp socket", "error", err)
}
return nil
return err
}
+20 -6
View File
@@ -10,6 +10,7 @@ import (
"net/netip"
"os"
"sync"
"sync/atomic"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/header"
@@ -64,7 +65,9 @@ func acquirePacket() *Packet {
}
type TesterConn struct {
Addr netip.AddrPort
// addr is read by nebula's own goroutines on every send and by the router's flow renderer, and a test can
// move it mid-run to simulate roaming, so it is atomic rather than a plain field.
addr atomic.Pointer[netip.AddrPort]
RxPackets chan *Packet // Packets to receive into nebula
TxPackets chan *Packet // Packets transmitted outside by nebula
@@ -82,13 +85,24 @@ type TesterConn struct {
}
func NewListener(l *slog.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn, error) {
return &TesterConn{
Addr: netip.AddrPortFrom(ip, uint16(port)),
c := &TesterConn{
RxPackets: make(chan *Packet, 10),
TxPackets: make(chan *Packet, 10),
done: make(chan struct{}),
l: l,
}, nil
}
c.SetAddr(netip.AddrPortFrom(ip, uint16(port)))
return c, nil
}
// GetAddr returns the underlay address this conn currently sends from.
func (u *TesterConn) GetAddr() netip.AddrPort {
return *u.addr.Load()
}
// SetAddr moves this conn to a new underlay address, standing in for a host waking up on a different network.
func (u *TesterConn) SetAddr(addr netip.AddrPort) {
u.addr.Store(&addr)
}
// Send will place a UdpPacket onto the receive queue for nebula to consume
@@ -147,7 +161,7 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
p.Data = p.Data[:len(b)]
}
copy(p.Data, b)
p.From = u.Addr
p.From = u.GetAddr()
p.To = addr
select {
case <-u.done:
@@ -178,7 +192,7 @@ func NewUDPStatsEmitter(_ []Conn) func() {
}
func (u *TesterConn) LocalAddr() (netip.AddrPort, error) {
return u.Addr, nil
return u.GetAddr(), nil
}
func (u *TesterConn) SupportsMultipleReaders() bool {
-29
View File
@@ -1,29 +0,0 @@
//go:build darwin
package nebula
import (
"time"
"golang.org/x/sys/unix"
)
// suspendClockDelta returns CLOCK_MONOTONIC - CLOCK_UPTIME_RAW. On macOS CLOCK_MONOTONIC keeps counting across
// system sleep while CLOCK_UPTIME_RAW (mach_absolute_time) pauses, so the difference grows by time spent asleep.
//
// The pausing clock is read first so scheduling jitter between the two reads biases the delta positive; the
// wakeDetector clamps out the noise.
//
// Caveat: on Apple Silicon the hardware timebase keeps ticking through sleep, which can make both clocks advance
// and the spread stay flat, leaving this detector blind. That fails safe (no clears, behavior as before); IOKit
// power notifications are the follow-up for full coverage on those machines.
func suspendClockDelta() (time.Duration, bool) {
var uptime, mono unix.Timespec
if err := unix.ClockGettime(unix.CLOCK_UPTIME_RAW, &uptime); err != nil {
return 0, false
}
if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &mono); err != nil {
return 0, false
}
return time.Duration(mono.Nano() - uptime.Nano()), true
}
-11
View File
@@ -1,11 +0,0 @@
//go:build !linux && !darwin && !windows
package nebula
import "time"
// suspendClockDelta reports that this platform has no usable clock pair for detecting system sleep; the wake
// detector stays dormant and dead tunnels are left to the normal traffic checks.
func suspendClockDelta() (time.Duration, bool) {
return 0, false
}
-26
View File
@@ -1,26 +0,0 @@
//go:build linux
package nebula
import (
"time"
"golang.org/x/sys/unix"
)
// suspendClockDelta returns CLOCK_BOOTTIME - CLOCK_MONOTONIC. CLOCK_MONOTONIC pauses while the system is suspended
// and CLOCK_BOOTTIME does not, so the difference only ever grows, and only by time spent suspended. Both reads are
// vDSO calls, cheap enough for a hot ticker.
//
// The pausing clock is read first so scheduling jitter between the two reads biases the delta positive; the
// wakeDetector clamps out the noise.
func suspendClockDelta() (time.Duration, bool) {
var mono, boot unix.Timespec
if err := unix.ClockGettime(unix.CLOCK_MONOTONIC, &mono); err != nil {
return 0, false
}
if err := unix.ClockGettime(unix.CLOCK_BOOTTIME, &boot); err != nil {
return 0, false
}
return time.Duration(boot.Nano() - mono.Nano()), true
}
-41
View File
@@ -1,41 +0,0 @@
//go:build windows
package nebula
import (
"sync"
"time"
"unsafe"
"golang.org/x/sys/windows"
)
var (
procQueryInterruptTime = windows.NewLazySystemDLL("kernelbase.dll").NewProc("QueryInterruptTime")
procQueryUnbiasedInterruptTime = windows.NewLazySystemDLL("kernel32.dll").NewProc("QueryUnbiasedInterruptTime")
// QueryInterruptTime needs Windows 10; probe once and stay dormant on anything older.
wakeClockAvailable = sync.OnceValue(func() bool {
return procQueryInterruptTime.Find() == nil && procQueryUnbiasedInterruptTime.Find() == nil
})
)
// suspendClockDelta returns interrupt time minus unbiased interrupt time, both in 100ns units. The unbiased count
// excludes time the system spends suspended while the biased one includes it, so the difference grows by exactly
// the time spent asleep.
//
// The pausing (unbiased) clock is read first so scheduling jitter between the two reads biases the delta positive;
// the wakeDetector clamps out the noise.
func suspendClockDelta() (time.Duration, bool) {
if !wakeClockAvailable() {
return 0, false
}
var unbiased, biased uint64
if r1, _, _ := procQueryUnbiasedInterruptTime.Call(uintptr(unsafe.Pointer(&unbiased))); r1 == 0 {
return 0, false
}
// Returns void, cannot fail once resolved.
_, _, _ = procQueryInterruptTime.Call(uintptr(unsafe.Pointer(&biased)))
return time.Duration(int64(biased-unbiased)) * 100, true
}
-47
View File
@@ -1,47 +0,0 @@
package nebula
import "time"
// wakeDetector notices when the machine has returned from system sleep and measures how long it was suspended.
//
// It samples the spread between two kernel clocks: one that pauses across a suspend and one that keeps counting
// (suspendClockDelta, per platform). While the machine is awake the spread is constant no matter how starved,
// stopped, or stepped this process is — SIGSTOP, debugger pauses, scheduler starvation, and NTP adjustments move
// both clocks together or neither, so none of them can fake a wake. A true suspend is the only thing that grows
// the spread, and it grows by exactly the time spent suspended.
//
// Sample is intended to piggyback on a ticker the caller already runs; it costs two clock reads. It is not safe
// for concurrent use.
type wakeDetector struct {
// read returns the current spread between the two clocks, false if this platform can't provide one.
read func() (time.Duration, bool)
last time.Duration
primed bool
}
func newWakeDetector() *wakeDetector {
return &wakeDetector{read: suspendClockDelta}
}
// Sample returns how long the machine was suspended since the previous call, 0 if it wasn't, and false if the
// platform has no way to tell. The first call primes the baseline and always reports 0.
func (w *wakeDetector) Sample() (time.Duration, bool) {
delta, ok := w.read()
if !ok {
return 0, false
}
if !w.primed {
w.primed = true
w.last = delta
return 0, true
}
slept := delta - w.last
w.last = delta
if slept < 0 {
// The clock pair is read non-atomically so tiny negative jitter is possible; it is never a wake.
slept = 0
}
return slept, true
}
-67
View File
@@ -1,67 +0,0 @@
package nebula
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func TestWakeDetector(t *testing.T) {
delta := time.Duration(0)
ok := true
w := &wakeDetector{read: func() (time.Duration, bool) { return delta, ok }}
// The first sample primes the baseline and never reports a wake, even with a pre-existing spread
delta = 3 * time.Hour
slept, sok := w.Sample()
assert.True(t, sok)
assert.Equal(t, time.Duration(0), slept)
// A stable spread means the machine never slept
slept, sok = w.Sample()
assert.True(t, sok)
assert.Equal(t, time.Duration(0), slept)
// The spread grows by exactly the time spent suspended
delta += 42 * time.Second
slept, sok = w.Sample()
assert.True(t, sok)
assert.Equal(t, 42*time.Second, slept)
// A wake is reported once, then the baseline moves with it
slept, sok = w.Sample()
assert.True(t, sok)
assert.Equal(t, time.Duration(0), slept)
// Negative jitter from the non-atomic clock pair reads clamps to zero
delta -= time.Microsecond
slept, sok = w.Sample()
assert.True(t, sok)
assert.Equal(t, time.Duration(0), slept)
// Consecutive suspends both report; the clamped jitter moved the baseline so it is not double-counted
delta += time.Minute
slept, _ = w.Sample()
assert.Equal(t, time.Minute, slept)
delta += time.Hour
slept, _ = w.Sample()
assert.Equal(t, time.Hour, slept)
// An unsupported platform read reports not-ok
ok = false
_, sok = w.Sample()
assert.False(t, sok)
}
// TestWakeDetectorPlatformClock smoke tests the real clock pair: two samples close together must not report a
// wake on a machine that isn't suspending mid-test.
func TestWakeDetectorPlatformClock(t *testing.T) {
w := newWakeDetector()
if _, ok := w.Sample(); !ok {
t.Skip("platform has no suspend clock pair")
}
slept, ok := w.Sample()
assert.True(t, ok)
assert.Less(t, slept, time.Second)
}