nbio for tun

Bumps [github.com/miekg/pkcs11](https://github.com/miekg/pkcs11) from 1.1.2-0.20231115102856-9078ad6b9d4b to 1.1.2.
- [Changelog](https://github.com/miekg/pkcs11/blob/master/release.go)
- [Commits](https://github.com/miekg/pkcs11/commits/v1.1.2)

---
updated-dependencies:
- dependency-name: github.com/miekg/pkcs11
  dependency-version: 1.1.2
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/release.yml

@@ -209,10 +209,11 @@ jobs:
         id: create_release
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_REF_NAME: ${{ github.ref_name }}
         run: |
           cd artifacts
           gh release create \
             --verify-tag \
-            --title "Release ${{ github.ref_name }}" \
-            "${{ github.ref_name }}" \
+            --title "Release ${GITHUB_REF_NAME}" \
+            "${GITHUB_REF_NAME}" \
             SHASUM256.txt *-latest/*.zip *-latest/*.tar.gz

.github/workflows/smoke/smoke-vagrant.sh

@@ -29,6 +29,17 @@ docker run --name lighthouse1 --rm "$CONTAINER" -config lighthouse1.yml -test
 docker run --name host2 --rm "$CONTAINER" -config host2.yml -test
 
 vagrant up
+
+# OpenBSD: synced folders are disabled because Vagrant's rsync installer
+# uses ftp.openbsd.org which no longer hosts packages for older releases.
+# Copy build artifacts in via scp instead.
+case "$1" in
+    openbsd-*)
+        vagrant ssh -c "sudo mkdir -p /nebula" -- -T
+        tar -cf - -C build . | vagrant ssh -c "sudo tar -xf - -C /nebula && sudo chmod -R a+r /nebula" -- -T
+        ;;
+esac
+
 vagrant ssh -c "cd /nebula && /nebula/$1-nebula -config host3.yml -test" -- -T
 
 docker run --name lighthouse1 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &

.github/workflows/smoke/smoke.sh

@@ -124,6 +124,7 @@ set -x
 # host2 speaking to host4 on UDP 4000 should allow it to reply, when firewall rules would normally not permit this
 docker exec host2 sh -c "/usr/bin/echo host2 | ncat -nuv 192.168.100.4 4000"
 docker exec host2 ncat -e '/usr/bin/echo helloagainfromhost2' -nkluv 0.0.0.0 4000 &
+sleep 1
 docker exec host4 sh -c "/usr/bin/echo host4 | ncat -nuv 192.168.100.2 4000"
 
 docker exec host4 sh -c 'kill 1'

.github/workflows/smoke/vagrant-openbsd-amd64/Vagrantfile

@@ -3,5 +3,5 @@
 Vagrant.configure("2") do |config|
   config.vm.box = "generic/openbsd7"
 
-  config.vm.synced_folder "../build", "/nebula", type: "rsync"
+  config.vm.synced_folder ".", "/vagrant", disabled: true
 end

CODEOWNERS

@@ -0,0 +1 @@
+#ECCN:Open Source

README.md

@@ -57,7 +57,7 @@ Check the [releases](https://github.com/slackhq/nebula/releases/latest) page for
     docker pull nebulaoss/nebula
     ```
 
-#### Mobile
+#### Mobile ([source code](https://github.com/DefinedNet/mobile_nebula))
 
 - [iOS](https://apps.apple.com/us/app/mobile-nebula/id1509587936?itsct=apps_box&itscg=30200)
 - [Android](https://play.google.com/store/apps/details?id=net.defined.mobile_nebula&pcampaignid=pcampaignidMKT-Other-global-all-co-prtnr-py-PartBadge-Mar2515-1)
@@ -76,6 +76,8 @@ Nebula was created to provide a mechanism for groups of hosts to communicate sec
 
 ## Getting started (quickly)
 
+**Don't want to manage your own PKI and lighthouses?** [Managed Nebula](https://www.defined.net/) from Defined Networking handles all of this for you.
+
 To set up a Nebula network, you'll need:
 
 #### 1. The [Nebula binaries](https://github.com/slackhq/nebula/releases) or [Distribution Packages](https://github.com/slackhq/nebula#distribution-packages) for your specific platform. Specifically you'll need `nebula-cert` and the specific nebula binary for each platform you use.

cert/ca_pool.go

@@ -1,11 +1,14 @@
 package cert
 
 import (
+	"bufio"
+	"bytes"
+	"encoding/pem"
 	"errors"
 	"fmt"
+	"io"
 	"net/netip"
 	"slices"
-	"strings"
 	"time"
 )
 
@@ -29,22 +32,46 @@ func NewCAPool() *CAPool {
 // If the pool contains any expired certificates, an ErrExpired will be
 // returned along with the pool. The caller must handle any such errors.
 func NewCAPoolFromPEM(caPEMs []byte) (*CAPool, error) {
+	return NewCAPoolFromPEMReader(bytes.NewReader(caPEMs))
+}
+
+// NewCAPoolFromPEMReader will create a new CA pool from the provided reader.
+// The reader must contain a PEM-encoded set of nebula certificates.
+func NewCAPoolFromPEMReader(r io.Reader) (*CAPool, error) {
 	pool := NewCAPool()
-	var err error
+
 	var expired bool
-	for {
-		caPEMs, err = pool.AddCAFromPEM(caPEMs)
-		if errors.Is(err, ErrExpired) {
-			expired = true
-			err = nil
+
+	scanner := bufio.NewScanner(r)
+	scanner.Split(SplitPEM)
+
+	for scanner.Scan() {
+		pemBytes := scanner.Bytes()
+
+		block, rest := pem.Decode(pemBytes)
+		if len(bytes.TrimSpace(rest)) > 0 {
+			return nil, ErrInvalidPEMBlock
 		}
+		if block == nil {
+			return nil, ErrInvalidPEMBlock
+		}
+
+		c, err := unmarshalCertificateBlock(block)
 		if err != nil {
 			return nil, err
 		}
-		if len(caPEMs) == 0 || strings.TrimSpace(string(caPEMs)) == "" {
-			break
+
+		err = pool.AddCA(c)
+		if errors.Is(err, ErrExpired) {
+			expired = true
+			continue
+		} else if err != nil {
+			return nil, err
 		}
 	}
+	if err := scanner.Err(); err != nil {
+		return nil, ErrInvalidPEMBlock
+	}
 
 	if expired {
 		return pool, ErrExpired

cert/ca_pool_test.go

@@ -1,7 +1,10 @@
 package cert
 
 import (
+	"bytes"
+	"io"
 	"net/netip"
+	"strings"
 	"testing"
 	"time"
 
@@ -112,6 +115,60 @@ k+coOv04r+zh33ISyhbsafnYduN17p2eD7CmHvHuerguXD9f32gcxo/KsFCKEjMe
 	assert.Len(t, ppppp.CAs, 1)
 }
 
+// oneByteReader wraps a reader to return at most 1 byte per Read call,
+// exercising the streaming accumulation logic in NewCAPoolFromPEMReader.
+type oneByteReader struct {
+	r io.Reader
+}
+
+func (o *oneByteReader) Read(p []byte) (int, error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+	return o.r.Read(p[:1])
+}
+
+func TestNewCAPoolFromPEMReader_EmptyReader(t *testing.T) {
+	pool, err := NewCAPoolFromPEMReader(bytes.NewReader(nil))
+	require.NoError(t, err)
+	assert.Empty(t, pool.CAs)
+
+	pool, err = NewCAPoolFromPEMReader(strings.NewReader("   \n\t\n  "))
+	require.NoError(t, err)
+	assert.Empty(t, pool.CAs)
+}
+
+func TestNewCAPoolFromPEMReader_OneByteReads(t *testing.T) {
+	ca1, _, _, pem1 := NewTestCaCert(Version2, Curve_CURVE25519, time.Now(), time.Now().Add(time.Hour), nil, nil, nil)
+	ca2, _, _, pem2 := NewTestCaCert(Version2, Curve_CURVE25519, time.Now(), time.Now().Add(time.Hour), nil, nil, nil)
+
+	bundle := append(pem1, pem2...)
+	pool, err := NewCAPoolFromPEMReader(&oneByteReader{r: bytes.NewReader(bundle)})
+	require.NoError(t, err)
+	assert.Len(t, pool.CAs, 2)
+
+	fp1, err := ca1.Fingerprint()
+	require.NoError(t, err)
+	fp2, err := ca2.Fingerprint()
+	require.NoError(t, err)
+
+	assert.Contains(t, pool.CAs, fp1)
+	assert.Contains(t, pool.CAs, fp2)
+}
+
+func TestNewCAPoolFromPEMReader_TruncatedPEM(t *testing.T) {
+	_, err := NewCAPoolFromPEMReader(strings.NewReader("-----BEGIN NEBULA CERTIFICATE-----\npartialdata"))
+	assert.ErrorIs(t, err, ErrInvalidPEMBlock)
+}
+
+func TestNewCAPoolFromPEMReader_TrailingGarbage(t *testing.T) {
+	_, _, _, pem1 := NewTestCaCert(Version2, Curve_CURVE25519, time.Now(), time.Now().Add(time.Hour), nil, nil, nil)
+
+	bundle := append(pem1, []byte("some trailing garbage")...)
+	_, err := NewCAPoolFromPEMReader(bytes.NewReader(bundle))
+	assert.ErrorIs(t, err, ErrInvalidPEMBlock)
+}
+
 func TestCertificateV1_Verify(t *testing.T) {
 	ca, _, caKey, _ := NewTestCaCert(Version1, Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, nil)
 	c, _, _, _ := NewTestCert(Version1, Curve_CURVE25519, ca, caKey, "test cert", time.Now(), time.Now().Add(5*time.Minute), nil, nil, nil)

cert/p256/p256.go

@@ -44,7 +44,12 @@ func swap(r, s []byte) ([]byte, []byte, error) {
 	}
 	sNormalized := nMod.Nat().Sub(bigS, nMod)
 
-	return r, sNormalized.Bytes(nMod), nil
+	result := sNormalized.Bytes(nMod)
+	for len(result) > 1 && result[0] == 0 {
+		result = result[1:]
+	}
+
+	return r, result, nil
 }
 
 func Normalize(sig []byte) ([]byte, error) {

cert/pem.go

@@ -1,12 +1,66 @@
 package cert
 
 import (
+	"bytes"
 	"encoding/pem"
+	"errors"
 	"fmt"
 
 	"golang.org/x/crypto/ed25519"
 )
 
+var ErrTruncatedPEMBlock = errors.New("truncated PEM block")
+
+// SplitPEM is a split function for bufio.Scanner that returns each PEM block.
+func SplitPEM(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	// Look for the start of a PEM block
+	start := bytes.Index(data, []byte("-----BEGIN "))
+	if start == -1 {
+		if atEOF && len(bytes.TrimSpace(data)) > 0 {
+			// Non-whitespace content with no PEM block
+			return 0, nil, ErrTruncatedPEMBlock
+		}
+		if atEOF {
+			return len(data), nil, nil
+		}
+		// Request more data
+		return 0, nil, nil
+	}
+
+	// Look for the end marker
+	endMarkerStart := bytes.Index(data[start:], []byte("-----END "))
+	if endMarkerStart == -1 {
+		if atEOF {
+			// Incomplete PEM block at EOF
+			return 0, nil, ErrTruncatedPEMBlock
+		}
+		// Need more data to find the end
+		return 0, nil, nil
+	}
+
+	// Find the actual end of the END line (after the newline)
+	endMarkerStart += start
+	endLineEnd := bytes.IndexByte(data[endMarkerStart:], '\n')
+	var end int
+	if endLineEnd == -1 {
+		if atEOF {
+			// END marker without newline at EOF - take it anyway
+			end = len(data)
+		} else {
+			// Need more data
+			return 0, nil, nil
+		}
+	} else {
+		end = endMarkerStart + endLineEnd + 1
+	}
+
+	// Extract the PEM block
+	pemBlock := data[start:end]
+
+	// Return the valid PEM block
+	return end, pemBlock, nil
+}
+
 const ( //cert banners
 	CertificateBanner   = "NEBULA CERTIFICATE"
 	CertificateV2Banner = "NEBULA CERTIFICATE V2"
@@ -37,19 +91,7 @@ func UnmarshalCertificateFromPEM(b []byte) (Certificate, []byte, error) {
 		return nil, r, ErrInvalidPEMBlock
 	}
 
-	var c Certificate
-	var err error
-
-	switch p.Type {
-	// Implementations must validate the resulting certificate contains valid information
-	case CertificateBanner:
-		c, err = unmarshalCertificateV1(p.Bytes, nil)
-	case CertificateV2Banner:
-		c, err = unmarshalCertificateV2(p.Bytes, nil, Curve_CURVE25519)
-	default:
-		return nil, r, ErrInvalidPEMCertificateBanner
-	}
-
+	c, err := unmarshalCertificateBlock(p)
 	if err != nil {
 		return nil, r, err
 	}
@@ -58,6 +100,20 @@ func UnmarshalCertificateFromPEM(b []byte) (Certificate, []byte, error) {
 
 }
 
+// unmarshalCertificateBlock decodes a single PEM block into a certificate.
+// It expects a Nebula certificate banner and returns ErrInvalidPEMCertificateBanner otherwise.
+func unmarshalCertificateBlock(block *pem.Block) (Certificate, error) {
+	switch block.Type {
+	// Implementations must validate the resulting certificate contains valid information
+	case CertificateBanner:
+		return unmarshalCertificateV1(block.Bytes, nil)
+	case CertificateV2Banner:
+		return unmarshalCertificateV2(block.Bytes, nil, Curve_CURVE25519)
+	default:
+		return nil, ErrInvalidPEMCertificateBanner
+	}
+}
+
 func marshalCertPublicKeyToPEM(c Certificate) []byte {
 	if c.IsCA() {
 		return MarshalSigningPublicKeyToPEM(c.Curve(), c.PublicKey())

cert/pem_test.go

@@ -1,12 +1,88 @@
 package cert
 
 import (
+	"bufio"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func scanAll(t *testing.T, input string) ([]string, error) {
+	t.Helper()
+	scanner := bufio.NewScanner(strings.NewReader(input))
+	scanner.Split(SplitPEM)
+	var blocks []string
+	for scanner.Scan() {
+		blocks = append(blocks, scanner.Text())
+	}
+	return blocks, scanner.Err()
+}
+
+func TestSplitPEM_Single(t *testing.T) {
+	input := "-----BEGIN TEST-----\ndata\n-----END TEST-----\n"
+	blocks, err := scanAll(t, input)
+	require.NoError(t, err)
+	require.Len(t, blocks, 1)
+	require.Equal(t, input, blocks[0])
+}
+
+func TestSplitPEM_Multiple(t *testing.T) {
+	block1 := "-----BEGIN TEST-----\naaa\n-----END TEST-----\n"
+	block2 := "-----BEGIN TEST-----\nbbb\n-----END TEST-----\n"
+	blocks, err := scanAll(t, block1+block2)
+	require.NoError(t, err)
+	require.Len(t, blocks, 2)
+	require.Equal(t, block1, blocks[0])
+	require.Equal(t, block2, blocks[1])
+}
+
+func TestSplitPEM_CommentsAndWhitespaceBetweenBlocks(t *testing.T) {
+	input := "# comment\n\n-----BEGIN TEST-----\naaa\n-----END TEST-----\n\n# another comment\n\n-----BEGIN TEST-----\nbbb\n-----END TEST-----\n"
+	blocks, err := scanAll(t, input)
+	require.NoError(t, err)
+	require.Len(t, blocks, 2)
+}
+
+func TestSplitPEM_Empty(t *testing.T) {
+	blocks, err := scanAll(t, "")
+	require.NoError(t, err)
+	require.Empty(t, blocks)
+}
+
+func TestSplitPEM_WhitespaceOnly(t *testing.T) {
+	blocks, err := scanAll(t, "  \n\t\n  ")
+	require.NoError(t, err)
+	require.Empty(t, blocks)
+}
+
+func TestSplitPEM_TrailingGarbage(t *testing.T) {
+	input := "-----BEGIN TEST-----\ndata\n-----END TEST-----\ngarbage"
+	blocks, err := scanAll(t, input)
+	require.ErrorIs(t, err, ErrTruncatedPEMBlock)
+	require.Len(t, blocks, 1)
+}
+
+func TestSplitPEM_TruncatedBlock(t *testing.T) {
+	input := "-----BEGIN TEST-----\npartial data with no end"
+	_, err := scanAll(t, input)
+	require.ErrorIs(t, err, ErrTruncatedPEMBlock)
+}
+
+func TestSplitPEM_NoEndNewline(t *testing.T) {
+	input := "-----BEGIN TEST-----\ndata\n-----END TEST-----"
+	blocks, err := scanAll(t, input)
+	require.NoError(t, err)
+	require.Len(t, blocks, 1)
+	require.Equal(t, input, blocks[0])
+}
+
+func TestSplitPEM_GarbageOnly(t *testing.T) {
+	_, err := scanAll(t, "this is not PEM data")
+	require.ErrorIs(t, err, ErrTruncatedPEMBlock)
+}
+
 func TestUnmarshalCertificateFromPEM(t *testing.T) {
 	goodCert := []byte(`
 # A good cert

cmd/nebula-cert/verify.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"strings"
 	"time"
 
 	"github.com/slackhq/nebula/cert"
@@ -40,21 +39,15 @@ func verify(args []string, out io.Writer, errOut io.Writer) error {
 		return err
 	}
 
-	rawCACert, err := os.ReadFile(*vf.caPath)
+	caFile, err := os.Open(*vf.caPath)
 	if err != nil {
 		return fmt.Errorf("error while reading ca: %w", err)
 	}
+	defer caFile.Close()
 
-	caPool := cert.NewCAPool()
-	for {
-		rawCACert, err = caPool.AddCAFromPEM(rawCACert)
-		if err != nil {
-			return fmt.Errorf("error while adding ca cert to pool: %w", err)
-		}
-
-		if rawCACert == nil || len(rawCACert) == 0 || strings.TrimSpace(string(rawCACert)) == "" {
-			break
-		}
+	caPool, err := cert.NewCAPoolFromPEMReader(caFile)
+	if err != nil && !errors.Is(err, cert.ErrExpired) {
+		return fmt.Errorf("error while adding ca cert to pool: %w", err)
 	}
 
 	rawCert, err := os.ReadFile(*vf.certPath)

cmd/nebula-cert/verify_test.go

@@ -64,7 +64,7 @@ func Test_verify(t *testing.T) {
 	err = verify([]string{"-ca", caFile.Name(), "-crt", "does_not_exist"}, ob, eb)
 	assert.Empty(t, ob.String())
 	assert.Empty(t, eb.String())
-	require.EqualError(t, err, "error while adding ca cert to pool: input did not contain a valid PEM encoded block")
+	require.ErrorIs(t, err, cert.ErrInvalidPEMBlock)
 
 	// make a ca for later
 	caPub, caPriv, _ := ed25519.GenerateKey(rand.Reader)

cmd/nebula-service/main.go

@@ -78,8 +78,16 @@ func main() {
 	}
 
 	if !*configTest {
-		ctrl.Start()
-		ctrl.ShutdownBlock()
+		wait, err := ctrl.Start()
+		if err != nil {
+			util.LogWithContextIfNeeded("Error while running", err, l)
+			os.Exit(1)
+		}
+
+		go ctrl.ShutdownBlock()
+		wait()
+
+		l.Info("Goodbye")
 	}
 
 	os.Exit(0)

cmd/nebula/main.go

@@ -72,9 +72,17 @@ func main() {
 	}
 
 	if !*configTest {
-		ctrl.Start()
+		wait, err := ctrl.Start()
+		if err != nil {
+			util.LogWithContextIfNeeded("Error while running", err, l)
+			os.Exit(1)
+		}
+
+		go ctrl.ShutdownBlock()
 		notifyReady(l)
-		ctrl.ShutdownBlock()
+		wait()
+
+		l.Info("Goodbye")
 	}
 
 	os.Exit(0)

control.go

@@ -2,9 +2,11 @@ package nebula
 
 import (
 	"context"
+	"errors"
 	"net/netip"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 
 	"github.com/sirupsen/logrus"
@@ -13,6 +15,16 @@ import (
 	"github.com/slackhq/nebula/overlay"
 )
 
+type RunState int
+
+const (
+	Stopped  RunState = 0 // The control has yet to be started
+	Started  RunState = 1 // The control has been started
+	Stopping RunState = 2 // The control is stopping
+)
+
+var ErrAlreadyStarted = errors.New("nebula is already started")
+
 // Every interaction here needs to take extra care to copy memory and not return or use arguments "as is" when touching
 // core. This means copying IP objects, slices, de-referencing pointers and taking the actual value, etc
 
@@ -26,6 +38,9 @@ type controlHostLister interface {
 }
 
 type Control struct {
+	stateLock sync.Mutex
+	state     RunState
+
 	f                      *Interface
 	l                      *logrus.Logger
 	ctx                    context.Context
@@ -49,10 +64,21 @@ type ControlHostInfo struct {
 	CurrentRelaysThroughMe []netip.Addr     `json:"currentRelaysThroughMe"`
 }
 
-// Start actually runs nebula, this is a nonblocking call. To block use Control.ShutdownBlock()
-func (c *Control) Start() {
+// Start actually runs nebula, this is a nonblocking call.
+// The returned function can be used to wait for nebula to fully stop.
+func (c *Control) Start() (func(), error) {
+	c.stateLock.Lock()
+	if c.state != Stopped {
+		c.stateLock.Unlock()
+		return nil, ErrAlreadyStarted
+	}
+
 	// Activate the interface
-	c.f.activate()
+	err := c.f.activate()
+	if err != nil {
+		c.stateLock.Unlock()
+		return nil, err
+	}
 
 	// Call all the delayed funcs that waited patiently for the interface to be created.
 	if c.sshStart != nil {
@@ -72,15 +98,33 @@ func (c *Control) Start() {
 	}
 
 	// Start reading packets.
-	c.f.run()
+	c.state = Started
+	c.stateLock.Unlock()
+	return c.f.run()
+}
+
+func (c *Control) State() RunState {
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	return c.state
 }
 
 func (c *Control) Context() context.Context {
 	return c.ctx
 }
 
-// Stop signals nebula to shutdown and close all tunnels, returns after the shutdown is complete
+// Stop is a non-blocking call that signals nebula to close all tunnels and shut down
 func (c *Control) Stop() {
+	c.stateLock.Lock()
+	if c.state != Started {
+		c.stateLock.Unlock()
+		// We are stopping or stopped already
+		return
+	}
+
+	c.state = Stopping
+	c.stateLock.Unlock()
+
 	// Stop the handshakeManager (and other services), to prevent new tunnels from
 	// being created while we're shutting them all down.
 	c.cancel()
@@ -89,7 +133,9 @@ func (c *Control) Stop() {
 	if err := c.f.Close(); err != nil {
 		c.l.WithError(err).Error("Close interface failed")
 	}
-	c.l.Info("Goodbye")
+	c.stateLock.Lock()
+	c.state = Stopped
+	c.stateLock.Unlock()
 }
 
 // ShutdownBlock will listen for and block on term and interrupt signals, calling Control.Stop() once signalled

dist/wireshark/nebula.lua

@@ -84,30 +84,24 @@ end
 
 function nebula.prefs_changed()
     if default_settings.all_ports == nebula.prefs.all_ports and default_settings.port == nebula.prefs.port then
-        -- Nothing changed, bail
         return
     end
 
-    -- Remove our old dissector
+    -- Remove all existing registrations
     DissectorTable.get("udp.port"):remove_all(nebula)
 
-    if nebula.prefs.all_ports and default_settings.all_ports ~= nebula.prefs.all_ports then
-        default_settings.all_port = nebula.prefs.all_ports
-
+    if nebula.prefs.all_ports then
+        -- Register on every port for hole punch capture
         for i=0, 65535 do
             DissectorTable.get("udp.port"):add(i, nebula)
         end
-
-        -- no need to establish again on specific ports
-        return
+    else
+        -- Register on the configured port only
+        DissectorTable.get("udp.port"):add(nebula.prefs.port, nebula)
     end
 
-
-    if default_settings.all_ports ~= nebula.prefs.all_ports then
-        -- Add our new port dissector
-        default_settings.port = nebula.prefs.port
-        DissectorTable.get("udp.port"):add(default_settings.port, nebula)
-    end
+    default_settings.all_ports = nebula.prefs.all_ports
+    default_settings.port = nebula.prefs.port
 end
 
 DissectorTable.get("udp.port"):add(default_settings.port, nebula)

e2e/handshake_manager_test.go

@@ -0,0 +1,565 @@
+//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"
+)
+
+// makeHandshakePacket creates a handshake packet with the given parameters.
+func makeHandshakePacket(from, to netip.AddrPort, subtype header.MessageSubType, remoteIndex uint32, counter uint64) *udp.Packet {
+	data := make([]byte, 200)
+	header.Encode(data, header.Version, header.Handshake, subtype, remoteIndex, counter)
+	for i := header.Len; i < len(data); i++ {
+		data[i] = byte(i)
+	}
+	return &udp.Packet{To: to, From: from, Data: data}
+}
+
+func TestHandshakeRetransmitDuplicate(t *testing.T) {
+	// Verify the responder correctly handles receiving the same msg1 multiple times
+	// (retransmission). The duplicate goes through CheckAndComplete -> ErrAlreadySeen
+	// and the cached response is resent.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+	defer r.RenderFlow()
+
+	t.Log("Trigger handshake from me to them")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+
+	t.Log("Grab my msg1")
+	msg1 := myControl.GetFromUDP(true)
+
+	t.Log("Inject msg1 into them, first time")
+	theirControl.InjectUDPPacket(msg1)
+	_ = theirControl.GetFromUDP(true)
+
+	t.Log("Inject the SAME msg1 again, tests ErrAlreadySeen path")
+	theirControl.InjectUDPPacket(msg1)
+	resp2 := theirControl.GetFromUDP(true)
+	assert.NotNil(t, resp2, "should get cached response on duplicate msg1")
+
+	t.Log("Complete handshake with cached response")
+	myControl.InjectUDPPacket(resp2)
+	myControl.WaitForType(1, 0, theirControl)
+
+	t.Log("Drain cached packet and verify tunnel works")
+	cachedPacket := theirControl.GetFromTun(true)
+	assertUdpPacket(t, []byte("Hi"), cachedPacket, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), 80, 80)
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	t.Log("Verify only one tunnel exists on each side")
+	assert.Len(t, myControl.ListHostmapHosts(false), 1)
+	assert.Len(t, theirControl.ListHostmapHosts(false), 1)
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeTruncatedPacketRecovery(t *testing.T) {
+	// Verify that a truncated handshake packet is ignored and the real
+	// packet can still complete the handshake.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+	defer r.RenderFlow()
+
+	t.Log("Trigger handshake")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+
+	t.Log("Get msg1 and deliver to responder")
+	msg1 := myControl.GetFromUDP(true)
+	theirControl.InjectUDPPacket(msg1)
+
+	t.Log("Get the real response")
+	realResp := theirControl.GetFromUDP(true)
+
+	t.Log("Truncate the response and inject, should be ignored")
+	truncResp := realResp.Copy()
+	truncResp.Data = truncResp.Data[:header.Len]
+	myControl.InjectUDPPacket(truncResp)
+
+	t.Log("Verify pending handshake survived the truncated packet")
+	assert.NotEmpty(t, myControl.ListHostmapHosts(true), "pending handshake should still exist")
+
+	t.Log("Inject real response, should complete handshake")
+	myControl.InjectUDPPacket(realResp)
+	myControl.WaitForType(1, 0, theirControl)
+
+	t.Log("Drain and verify tunnel")
+	cachedPacket := theirControl.GetFromTun(true)
+	assertUdpPacket(t, []byte("Hi"), cachedPacket, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), 80, 80)
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeOrphanedMsg2Dropped(t *testing.T) {
+	// A msg2 arriving with no matching pending index should be silently dropped
+	// with no response sent and no state changes.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+	defer r.RenderFlow()
+
+	t.Log("Complete a normal handshake")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+	r.RouteForAllUntilTxTun(theirControl)
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	t.Log("Record hostmap state")
+	myIndexes := len(myControl.ListHostmapIndexes(false))
+
+	t.Log("Inject a fake msg2 with unknown RemoteIndex")
+	myControl.InjectUDPPacket(makeHandshakePacket(theirUdpAddr, myUdpAddr, header.HandshakeIXPSK0, 0xDEADBEEF, 2))
+
+	t.Log("Verify no new indexes created")
+	assert.Equal(t, myIndexes, len(myControl.ListHostmapIndexes(false)))
+
+	t.Log("Verify no UDP response was sent")
+	time.Sleep(100 * time.Millisecond)
+	assert.Nil(t, myControl.GetFromUDP(false), "should not send a response to orphaned msg2")
+
+	t.Log("Verify existing tunnel still works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeUnknownMessageCounter(t *testing.T) {
+	// A handshake packet with an unexpected message counter should be silently
+	// dropped with no side effects and no UDP response.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, _, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	t.Log("Inject handshake with MessageCounter=3")
+	myControl.InjectUDPPacket(makeHandshakePacket(theirUdpAddr, myUdpAddr, header.HandshakeIXPSK0, 0, 3))
+
+	t.Log("Inject handshake with MessageCounter=99")
+	myControl.InjectUDPPacket(makeHandshakePacket(theirUdpAddr, myUdpAddr, header.HandshakeIXPSK0, 0, 99))
+
+	t.Log("Verify no tunnels or pending handshakes")
+	assert.Empty(t, myControl.ListHostmapHosts(false))
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+
+	t.Log("Verify no UDP response was sent")
+	time.Sleep(100 * time.Millisecond)
+	assert.Nil(t, myControl.GetFromUDP(false))
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeUnknownSubtype(t *testing.T) {
+	// A handshake packet with an unknown subtype should be silently dropped.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, _, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, _, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.Start()
+	theirControl.Start()
+
+	t.Log("Inject handshake with unknown subtype 99")
+	myControl.InjectUDPPacket(makeHandshakePacket(theirUdpAddr, myUdpAddr, header.MessageSubType(99), 0, 1))
+
+	t.Log("Verify no tunnels or pending handshakes")
+	assert.Empty(t, myControl.ListHostmapHosts(false))
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+
+	t.Log("Verify no UDP response was sent")
+	time.Sleep(100 * time.Millisecond)
+	assert.Nil(t, myControl.GetFromUDP(false))
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeLateResponse(t *testing.T) {
+	// After a handshake times out, a late response should be silently ignored
+	// with no new tunnels created.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, _, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{
+		"handshakes": m{
+			"try_interval": "200ms",
+			"retries":      2,
+		},
+	})
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	t.Log("Trigger handshake from me")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+
+	t.Log("Grab msg1 but don't deliver")
+	msg1 := myControl.GetFromUDP(true)
+
+	t.Log("Wait for handshake to time out")
+	for i := 0; i < 5; i++ {
+		time.Sleep(300 * time.Millisecond)
+		myControl.GetFromUDP(false)
+	}
+
+	t.Log("Confirm no pending handshakes remain")
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+
+	t.Log("Deliver old msg1 to them, they create a tunnel")
+	theirControl.InjectUDPPacket(msg1)
+	resp := theirControl.GetFromUDP(true)
+	assert.NotNil(t, resp)
+
+	t.Log("Inject late response into me, should be ignored")
+	myControl.InjectUDPPacket(resp)
+
+	t.Log("No tunnel should exist on my side")
+	assert.Empty(t, myControl.ListHostmapHosts(false))
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeSelfConnectionRejected(t *testing.T) {
+	// Verify that a node rejects a handshake containing its own VPN IP in the
+	// peer cert. We do this by sending the initiator's own msg1 back to itself.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+
+	// Need a lighthouse entry to trigger a handshake
+	myControl.InjectLightHouseAddr(netip.MustParseAddr("10.128.0.2"), netip.MustParseAddrPort("10.0.0.2:4242"))
+
+	myControl.Start()
+
+	t.Log("Trigger handshake from me")
+	myControl.InjectTunUDPPacket(netip.MustParseAddr("10.128.0.2"), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+	msg1 := myControl.GetFromUDP(true)
+
+	t.Log("Drain any handshake retransmits before injecting")
+	time.Sleep(100 * time.Millisecond)
+	for myControl.GetFromUDP(false) != nil {
+	}
+
+	t.Log("Feed my own msg1 back to me as if it came from someone else")
+	selfMsg := msg1.Copy()
+	selfMsg.From = netip.MustParseAddrPort("10.0.0.99:4242")
+	selfMsg.To = myUdpAddr
+	myControl.InjectUDPPacket(selfMsg)
+
+	t.Log("Verify no response was sent (self-connection rejected)")
+	time.Sleep(100 * time.Millisecond)
+	// Drain any further retransmits from the original handshake, then check
+	// that none of them are a handshake response (MessageCounter=2)
+	h := &header.H{}
+	for {
+		p := myControl.GetFromUDP(false)
+		if p == nil {
+			break
+		}
+		_ = h.Parse(p.Data)
+		assert.NotEqual(t, uint64(2), h.MessageCounter,
+			"should not send a stage 2 response to self-connection")
+	}
+
+	t.Log("Verify no tunnel to myself was created")
+	assert.Nil(t, myControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false))
+
+	myControl.Stop()
+}
+
+func TestHandshakeMessageCounter0Dropped(t *testing.T) {
+	// MessageCounter=0 is not a valid handshake message and should be dropped.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, _, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	_, _, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.Start()
+
+	t.Log("Inject handshake with MessageCounter=0")
+	myControl.InjectUDPPacket(makeHandshakePacket(theirUdpAddr, myUdpAddr, header.HandshakeIXPSK0, 0, 0))
+
+	time.Sleep(100 * time.Millisecond)
+	assert.Empty(t, myControl.ListHostmapHosts(false))
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+	assert.Nil(t, myControl.GetFromUDP(false))
+
+	myControl.Stop()
+}
+
+func TestHandshakeRemoteAllowList(t *testing.T) {
+	// Verify that a handshake from a blocked underlay IP is dropped with no
+	// response and no state changes. Then verify the same packet from an
+	// allowed IP succeeds.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{
+		"lighthouse": m{
+			"remote_allow_list": m{
+				"10.0.0.0/8": true,
+				"0.0.0.0/0":  false,
+			},
+		},
+	})
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+	defer r.RenderFlow()
+
+	t.Log("Trigger handshake from them")
+	theirControl.InjectTunUDPPacket(myVpnIpNet[0].Addr(), 80, theirVpnIpNet[0].Addr(), 80, []byte("Hi"))
+	msg1 := theirControl.GetFromUDP(true)
+
+	t.Log("Rewrite the source to a blocked IP and inject")
+	blockedMsg := msg1.Copy()
+	blockedMsg.From = netip.MustParseAddrPort("192.168.1.1:4242")
+	myControl.InjectUDPPacket(blockedMsg)
+
+	t.Log("Verify no tunnel, no pending, no response from blocked source")
+	time.Sleep(100 * time.Millisecond)
+	assert.Empty(t, myControl.ListHostmapHosts(false))
+	assert.Empty(t, myControl.ListHostmapHosts(true))
+	assert.Nil(t, myControl.GetFromUDP(false), "should not respond to blocked source")
+
+	t.Log("Now inject the real packet from the allowed source")
+	myControl.InjectUDPPacket(msg1)
+
+	t.Log("Verify handshake completes from allowed source")
+	resp := myControl.GetFromUDP(true)
+	assert.NotNil(t, resp)
+	theirControl.InjectUDPPacket(resp)
+	theirControl.WaitForType(1, 0, myControl)
+
+	t.Log("Drain cached packet and verify tunnel works")
+	cachedPacket := myControl.GetFromTun(true)
+	assertUdpPacket(t, []byte("Hi"), cachedPacket, theirVpnIpNet[0].Addr(), myVpnIpNet[0].Addr(), 80, 80)
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeAlreadySeenPreferredRemote(t *testing.T) {
+	// When a duplicate msg1 arrives via ErrAlreadySeen, verify the tunnel
+	// remains functional and hostmap index count is stable.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+	defer r.RenderFlow()
+
+	t.Log("Complete a normal handshake via the router")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi"))
+	r.RouteForAllUntilTxTun(theirControl)
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	t.Log("Record hostmap state")
+	theirIndexes := len(theirControl.ListHostmapIndexes(false))
+	hi := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	assert.NotNil(t, hi)
+	originalRemote := hi.CurrentRemote
+
+	t.Log("Re-trigger traffic to cause a new handshake attempt (ErrAlreadySeen)")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("roam"))
+	r.RouteForAllUntilTxTun(theirControl)
+
+	t.Log("Verify tunnel still works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	t.Log("Verify remote is still valid and index count is stable")
+	hi2 := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	assert.NotNil(t, hi2)
+	assert.Equal(t, originalRemote, hi2.CurrentRemote)
+	assert.Equal(t, theirIndexes, len(theirControl.ListHostmapIndexes(false)),
+		"no extra indexes should be created from ErrAlreadySeen")
+
+	myControl.Stop()
+	theirControl.Stop()
+}
+
+func TestHandshakeWrongResponderPacketStore(t *testing.T) {
+	// Verify that when the wrong host responds, the cached packets are
+	// transferred to the new handshake, the evil tunnel is closed, evil's
+	// address is blocked, and the correct tunnel is eventually established.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.100/24", nil)
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.99/24", nil)
+	evilControl, evilVpnIpNet, evilUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "evil", "10.128.0.2/24", nil)
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), evilUdpAddr)
+
+	r := router.NewR(t, myControl, theirControl, evilControl)
+	defer r.RenderFlow()
+
+	myControl.Start()
+	theirControl.Start()
+	evilControl.Start()
+
+	t.Log("Send multiple packets to them (cached during handshake)")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("packet1"))
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("packet2"))
+
+	t.Log("Route until evil tunnel is closed")
+	h := &header.H{}
+	r.RouteForAllExitFunc(func(p *udp.Packet, c *nebula.Control) router.ExitType {
+		if err := h.Parse(p.Data); err != nil {
+			panic(err)
+		}
+		if h.Type == header.CloseTunnel && p.To == evilUdpAddr {
+			return router.RouteAndExit
+		}
+		return router.KeepRouting
+	})
+
+	t.Log("Verify evil's address is blocked in the new pending handshake")
+	pendingHI := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), true)
+	if pendingHI != nil {
+		assert.NotContains(t, pendingHI.RemoteAddrs, evilUdpAddr,
+			"evil's address should be blocked")
+	}
+
+	t.Log("Inject correct lighthouse addr for them")
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+
+	t.Log("Route until cached packets arrive at the real them")
+	p := r.RouteForAllUntilTxTun(theirControl)
+	assert.NotNil(t, p, "a cached packet should be delivered to the correct host")
+
+	t.Log("Verify the correct host has a tunnel")
+	assertHostInfoPair(t, myUdpAddr, theirUdpAddr, myVpnIpNet, theirVpnIpNet, myControl, theirControl)
+
+	t.Log("Verify no hostinfo artifacts from evil remain")
+	assert.Nil(t, myControl.GetHostInfoByVpnAddr(evilVpnIpNet[0].Addr(), true),
+		"no pending hostinfo for evil")
+	assert.Nil(t, myControl.GetHostInfoByVpnAddr(evilVpnIpNet[0].Addr(), false),
+		"no main hostinfo for evil")
+
+	myControl.Stop()
+	theirControl.Stop()
+	evilControl.Stop()
+}
+
+func TestHandshakeRelayComplete(t *testing.T) {
+	// Verify that a relay handshake completes correctly and relay state is
+	// properly maintained on all three nodes.
+
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, _, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{"relay": m{"use_relays": true}})
+	relayControl, relayVpnIpNet, relayUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "relay", "10.128.0.128/24", m{"relay": m{"am_relay": true}})
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, 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()
+
+	t.Log("Trigger handshake via relay")
+	myControl.InjectTunUDPPacket(theirVpnIpNet[0].Addr(), 80, myVpnIpNet[0].Addr(), 80, []byte("Hi via relay"))
+
+	p := r.RouteForAllUntilTxTun(theirControl)
+	assertUdpPacket(t, []byte("Hi via relay"), p, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), 80, 80)
+
+	t.Log("Verify bidirectional tunnel via relay")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	t.Log("Verify relay state on my side shows relay-to-me")
+	myHI := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
+	assert.NotNil(t, myHI)
+	assert.NotEmpty(t, myHI.CurrentRelaysToMe, "should have relay-to-me for them")
+
+	t.Log("Verify relay state on their side shows relay-to-me")
+	theirHI := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	assert.NotNil(t, theirHI)
+	assert.NotEmpty(t, theirHI.CurrentRelaysToMe, "should have relay-to-me for me")
+
+	t.Log("Verify relay node shows through-me relays")
+	relayHI := relayControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	assert.NotNil(t, relayHI)
+
+	myControl.Stop()
+	relayControl.Stop()
+	theirControl.Stop()
+}
+
+// NOTE: Relay V1 cert + IPv6 rejection is not tested here because
+// InjectTunUDPPacket from a V4 node to a V6 address panics in the test
+// framework. The check is in handshake_manager.go handleOutbound relay
+// logic (lines ~304-313): if the relay host has a V1 cert and either
+// address is IPv6, the relay is skipped.
+
+// NOTE: Relay reestablishment (Disestablished state transition) is covered
+// by the existing TestReestablishRelays in handshakes_test.go.

e2e/tunnels_test.go

@@ -12,6 +12,8 @@ import (
 	"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"
 	"gopkg.in/yaml.v3"
 )
@@ -365,3 +367,106 @@ func TestCrossStackRelaysWork(t *testing.T) {
 	//theirControl.Stop()
 	//relayControl.Stop()
 }
+
+func TestCloseTunnelAuthenticated(t *testing.T) {
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "5s"}})
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "10m"}})
+
+	// Share our underlay information
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	// Start the servers
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+
+	r.Log("Assert the tunnel between me and them works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	r.Log("Close the tunnel")
+	myControl.CloseTunnel(theirVpnIpNet[0].Addr(), false)
+	r.FlushAll()
+
+	waitStart := time.Now()
+	for {
+		myIndexes := len(myControl.GetHostmap().Indexes)
+		theirIndexes := len(theirControl.GetHostmap().Indexes)
+		if myIndexes == 0 && theirIndexes == 0 {
+			break
+		}
+
+		since := time.Since(waitStart)
+		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
+		if since > time.Second*6 {
+			t.Fatal("Tunnel should have been declared inactive after 2 seconds and before 6 seconds")
+		}
+
+		time.Sleep(1 * time.Second)
+		//r.FlushAll()
+	}
+
+	r.Logf("Happy path success, tunnels were dropped within %v", time.Since(waitStart))
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+	r.Log("Assert another tunnel between me and them works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+	hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
+	if hi == nil {
+		t.Fatal("There is no hostinfo for this tunnel")
+	}
+	myHi := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	if myHi == nil {
+		t.Fatal("There is no hostinfo for my tunnel")
+	}
+	r.Log("It does")
+
+	buf := make([]byte, 1024)
+	hdr := header.H{
+		Version:        1,
+		Type:           header.CloseTunnel,
+		Subtype:        0,
+		Reserved:       0,
+		RemoteIndex:    hi.RemoteIndex,
+		MessageCounter: 5,
+	}
+	out, err := hdr.Encode(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pkt := &udp.Packet{
+		To:   hi.CurrentRemote,
+		From: myHi.CurrentRemote,
+		Data: out,
+	}
+	r.InjectUDPPacket(myControl, theirControl, pkt)
+	r.Log("Injected bogus close tunnel. Let's see!")
+	waitStart = time.Now()
+	for {
+		myIndexes := len(myControl.GetHostmap().Indexes)
+		theirIndexes := len(theirControl.GetHostmap().Indexes)
+		if myIndexes == 0 {
+			t.Fatal("myIndexes should not be 0")
+		}
+		if theirIndexes == 0 {
+			t.Fatal("theirIndexes should not be 0, they should have rejected this bogus packet")
+		}
+
+		since := time.Since(waitStart)
+		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
+		if since > time.Second*4 {
+			t.Log("The tunnel would have been gone by now")
+			break
+		}
+
+		time.Sleep(1 * time.Second)
+		r.FlushAll()
+	}
+
+	myControl.Stop()
+	theirControl.Stop()
+}

examples/config.yml

@@ -204,6 +204,12 @@ punchy:
   # Trusted SSH CA public keys. These are the public keys of the CAs that are allowed to sign SSH keys for access.
   #trusted_cas:
     #- "ssh public key string"
+  # sandbox_dir restricts file paths for profiling commands (start-cpu-profile, save-heap-profile,
+  # save-mutex-profile) to the specified directory. Relative paths will be resolved within this directory,
+  # and absolute paths outside of it will be rejected. Default is $TMP/nebula-debug.
+  # The directory is NOT automatically created.
+  # Overriding this to "" is the same as "/" and will allow overwriting any path on the host.
+  #sandbox_dir: /var/tmp/nebula-debug
 
 # EXPERIMENTAL: relay support for networks that can't establish direct connections.
 relay:

firewall.go

@@ -230,7 +230,7 @@ func NewFirewallFromConfig(l *logrus.Logger, cs *CertState, c *config.C) (*Firew
 	case "drop":
 		fw.OutSendReject = false
 	default:
-		l.WithField("action", inboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
+		l.WithField("action", outboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
 		fw.OutSendReject = false
 	}
 

go.mod

@@ -13,8 +13,8 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/gopacket v1.1.19
 	github.com/kardianos/service v1.2.4
-	github.com/miekg/dns v1.1.70
-	github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b
+	github.com/miekg/dns v1.1.72
+	github.com/miekg/pkcs11 v1.1.2
 	github.com/nbrownus/go-metrics-prometheus v0.0.0-20210712211119-974a6260965f
 	github.com/prometheus/client_golang v1.23.2
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475

go.sum

@@ -85,10 +85,10 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
-github.com/miekg/dns v1.1.70 h1:DZ4u2AV35VJxdD9Fo9fIWm119BsQL5cZU1cQ9s0LkqA=
-github.com/miekg/dns v1.1.70/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
-github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b h1:J/AzCvg5z0Hn1rqZUJjpbzALUmkKX0Zwbc/i4fw7Sfk=
-github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b/go.mod h1:XsNlhZGX73bx86s2hdc/FuaLm2CPZJemRLMA+WTFxgs=
+github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
+github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
+github.com/miekg/pkcs11 v1.1.2 h1:/VxmeAX5qU6Q3EwafypogwWbYryHFmF2RpkJmw3m4MQ=
+github.com/miekg/pkcs11 v1.1.2/go.mod h1:XsNlhZGX73bx86s2hdc/FuaLm2CPZJemRLMA+WTFxgs=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=

interface.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/netip"
-	"os"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -87,6 +87,7 @@ type Interface struct {
 
 	writers []udp.Conn
 	readers []io.ReadWriteCloser
+	wg      sync.WaitGroup
 
 	metricHandshakes    metrics.Histogram
 	messageMetrics      *MessageMetrics
@@ -209,7 +210,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 // activate creates the interface on the host. After the interface is created, any
 // other services that want to bind listeners to its IP may do so successfully. However,
 // the interface isn't going to process anything until run() is called.
-func (f *Interface) activate() {
+func (f *Interface) activate() error {
 	// actually turn on tun dev
 
 	addr, err := f.outside.LocalAddr()
@@ -237,28 +238,36 @@ func (f *Interface) activate() {
 		if i > 0 {
 			reader, err = f.inside.NewMultiQueueReader()
 			if err != nil {
-				f.l.Fatal(err)
+				return err
 			}
 		}
 		f.readers[i] = reader
 	}
 
-	if err := f.inside.Activate(); err != nil {
+	if err = f.inside.Activate(); err != nil {
 		f.inside.Close()
-		f.l.Fatal(err)
+		return err
 	}
+
+	return nil
 }
 
-func (f *Interface) run() {
+func (f *Interface) run() (func(), error) {
 	// Launch n queues to read packets from udp
 	for i := 0; i < f.routines; i++ {
-		go f.listenOut(i)
+		f.wg.Go(func() {
+			f.listenOut(i)
+		})
 	}
 
 	// Launch n queues to read packets from tun dev
 	for i := 0; i < f.routines; i++ {
-		go f.listenIn(f.readers[i], i)
+		f.wg.Go(func() {
+			f.listenIn(f.readers[i], i)
+		})
 	}
+
+	return f.wg.Wait, nil
 }
 
 func (f *Interface) listenOut(i int) {
@@ -276,9 +285,16 @@ func (f *Interface) listenOut(i int) {
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
 
-	li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
+	err := li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
 		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get(f.l))
 	})
+
+	if err != nil && !f.closed.Load() {
+		f.l.WithError(err).Error("Error while reading inbound packet, closing")
+		//TODO: Trigger Control to close
+	}
+
+	f.l.Infof("underlay reader %v is done", i)
 }
 
 func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
@@ -292,17 +308,17 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
 	for {
 		n, err := reader.Read(packet)
 		if err != nil {
-			if errors.Is(err, os.ErrClosed) && f.closed.Load() {
-				return
+			if !f.closed.Load() {
+				f.l.WithError(err).WithField("reader", i).Error("Error while reading outbound packet, closing")
+				//TODO: Trigger Control to close
 			}
-
-			f.l.WithError(err).Error("Error while reading outbound packet")
-			// This only seems to happen when something fatal happens to the fd, so exit.
-			os.Exit(2)
+			break
 		}
 
 		f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i, conntrackCache.Get(f.l))
 	}
+
+	f.l.Infof("overlay reader %v is done", i)
 }
 
 func (f *Interface) RegisterConfigChangeCallbacks(c *config.C) {
@@ -477,23 +493,26 @@ func (f *Interface) GetCertState() *CertState {
 }
 
 func (f *Interface) Close() error {
+	var err error
 	f.closed.Store(true)
 
-	for _, u := range f.writers {
-		err := u.Close()
+	// Release the udp readers
+	for i, u := range f.writers {
+		err = u.Close()
 		if err != nil {
-			f.l.WithError(err).Error("Error while closing udp socket")
-		}
-	}
-	for i, r := range f.readers {
-		if i == 0 {
-			continue // f.readers[0] is f.inside, which we want to save for last
-		}
-		if err := r.Close(); err != nil {
-			f.l.WithError(err).Error("Error while closing tun reader")
+			f.l.WithError(err).WithField("writer", i).Error("Error while closing udp socket")
+		}
+	}
+
+	// Release the tun readers
+	for i, r := range f.readers {
+		if i == 0 {
+			continue // f.readers[0] is f.inside, which we want to save for last, since it closes other stuff too
+		}
+		if err = r.Close(); err != nil {
+			f.l.WithError(err).WithField("reader", i).Error("Error while closing tun reader")
 		}
 	}
 
-	// Release the tun device
 	return f.inside.Close()
 }

main.go

@@ -288,15 +288,15 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 	}
 
 	return &Control{
-		ifce,
-		l,
-		ctx,
-		cancel,
-		sshStart,
-		statsStart,
-		dnsStart,
-		lightHouse.StartUpdateWorker,
-		connManager.Start,
+		f:                      ifce,
+		l:                      l,
+		ctx:                    ctx,
+		cancel:                 cancel,
+		sshStart:               sshStart,
+		statsStart:             statsStart,
+		dnsStart:               dnsStart,
+		lighthouseStart:        lightHouse.StartUpdateWorker,
+		connectionManagerStart: connManager.Start,
 	}, nil
 }
 

outside.go

@@ -190,6 +190,13 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 		if !f.handleEncrypted(ci, via, h) {
 			return
 		}
+		_, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb)
+		if err != nil {
+			hostinfo.logger(f.l).WithError(err).WithField("from", via).
+				WithField("packet", packet).
+				Error("Failed to decrypt CloseTunnel packet")
+			return
+		}
 
 		hostinfo.logger(f.l).WithField("from", via).
 			Info("Close tunnel received, tearing down.")

overlay/tun_linux.go

@@ -26,7 +26,6 @@ import (
 
 type tun struct {
 	io.ReadWriteCloser
-	fd          int
 	Device      string
 	vpnNetworks []netip.Prefix
 	MaxMTU      int
@@ -85,7 +84,7 @@ func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []net
 }
 
 func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) {
-	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
+	fd, err := unix.Open("/dev/net/tun", os.O_RDWR|unix.O_NONBLOCK, 0)
 	if err != nil {
 		// If /dev/net/tun doesn't exist, try to create it (will happen in docker)
 		if os.IsNotExist(err) {
@@ -98,7 +97,7 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
 				return nil, fmt.Errorf("failed to create /dev/net/tun: %w", err)
 			}
 
-			fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
+			fd, err = unix.Open("/dev/net/tun", os.O_RDWR|unix.O_NONBLOCK, 0)
 			if err != nil {
 				return nil, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
 			}
@@ -136,7 +135,6 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
 func newTunGeneric(c *config.C, l *logrus.Logger, file *os.File, vpnNetworks []netip.Prefix) (*tun, error) {
 	t := &tun{
 		ReadWriteCloser:           file,
-		fd:                        int(file.Fd()),
 		vpnNetworks:               vpnNetworks,
 		TXQueueLen:                c.GetInt("tun.tx_queue", 500),
 		useSystemRoutes:           c.GetBool("tun.use_system_route_table", false),
@@ -238,22 +236,6 @@ func (t *tun) SupportsMultiqueue() bool {
 	return true
 }
 
-type MultiQueueReader struct {
-	fd int
-}
-
-func (m *MultiQueueReader) Read(p []byte) (int, error) {
-	return unix.Read(m.fd, p)
-}
-
-func (m *MultiQueueReader) Close() error {
-	return unix.Close(m.fd)
-}
-
-func (m *MultiQueueReader) Write(p []byte) (int, error) {
-	return write(m.fd, p)
-}
-
 func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
 	if err != nil {
@@ -264,11 +246,12 @@ func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
 	copy(req.Name[:], t.Device)
 	if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
-		_ = unix.Close(fd)
 		return nil, err
 	}
 
-	return &MultiQueueReader{fd: fd}, nil
+	file := os.NewFile(uintptr(fd), "/dev/net/tun")
+
+	return file, nil
 }
 
 func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
@@ -276,37 +259,6 @@ func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
 	return r
 }
 
-func write(fd int, b []byte) (int, error) {
-	var nn int
-	maximum := len(b)
-
-	for {
-		n, err := unix.Write(fd, b[nn:maximum])
-		if n > 0 {
-			nn += n
-		}
-		if nn == len(b) {
-			return nn, err
-		}
-
-		if err != nil {
-			return nn, err
-		}
-
-		if n == 0 {
-			return nn, io.ErrUnexpectedEOF
-		}
-	}
-}
-
-func (t *tun) Read(p []byte) (int, error) {
-	return unix.Read(t.fd, p)
-}
-
-func (t *tun) Write(b []byte) (int, error) {
-	return write(t.fd, b)
-}
-
 func (t *tun) deviceBytes() (o [16]byte) {
 	for i, c := range t.Device {
 		o[i] = byte(c)
@@ -732,16 +684,24 @@ func (t *tun) updateRoutes(r netlink.RouteUpdate) {
 func (t *tun) Close() error {
 	if t.routeChan != nil {
 		close(t.routeChan)
-	}
-
-	if t.ReadWriteCloser != nil {
-		_ = t.ReadWriteCloser.Close()
+		t.routeChan = nil
 	}
 
 	if t.ioctlFd > 0 {
-		_ = os.NewFile(t.ioctlFd, "ioctlFd").Close()
+		err := os.NewFile(t.ioctlFd, "ioctlFd").Close()
+		if err != nil {
+			t.l.WithField("error", err).Error("Failed to close ioctl fd")
+		}
 		t.ioctlFd = 0
 	}
 
+	if t.ReadWriteCloser != nil {
+		err := t.ReadWriteCloser.Close()
+		if err != nil {
+			t.l.WithField("error", err).Error("Failed to close tun file")
+			return err
+		}
+	}
+
 	return nil
 }

pki.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/netip"
 	"os"
@@ -487,25 +488,25 @@ func loadCertificate(b []byte) (cert.Certificate, []byte, error) {
 }
 
 func loadCAPoolFromConfig(l *logrus.Logger, c *config.C) (*cert.CAPool, error) {
-	var rawCA []byte
-	var err error
-
 	caPathOrPEM := c.GetString("pki.ca", "")
 	if caPathOrPEM == "" {
 		return nil, errors.New("no pki.ca path or PEM data provided")
 	}
 
-	if strings.Contains(caPathOrPEM, "-----BEGIN") {
-		rawCA = []byte(caPathOrPEM)
+	var caReader io.ReadCloser
+	var err error
 
+	if strings.Contains(caPathOrPEM, "-----BEGIN") {
+		caReader = io.NopCloser(strings.NewReader(caPathOrPEM))
 	} else {
-		rawCA, err = os.ReadFile(caPathOrPEM)
+		caReader, err = os.Open(caPathOrPEM)
 		if err != nil {
 			return nil, fmt.Errorf("unable to read pki.ca file %s: %s", caPathOrPEM, err)
 		}
 	}
+	defer caReader.Close()
 
-	caPool, err := cert.NewCAPoolFromPEM(rawCA)
+	caPool, err := cert.NewCAPoolFromPEMReader(caReader)
 	if errors.Is(err, cert.ErrExpired) {
 		var expired int
 		for _, crt := range caPool.CAs {

pki_hup_benchmark_test.go

@@ -0,0 +1,121 @@
+package nebula
+
+import (
+	"bytes"
+	"fmt"
+	"net/netip"
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/slackhq/nebula/cert"
+	cert_test "github.com/slackhq/nebula/cert_test"
+	"github.com/slackhq/nebula/config"
+	"github.com/slackhq/nebula/test"
+	"github.com/stretchr/testify/require"
+)
+
+func BenchmarkReloadConfigWithCAs(b *testing.B) {
+	prevProcs := runtime.GOMAXPROCS(1)
+	b.Cleanup(func() { runtime.GOMAXPROCS(prevProcs) })
+
+	for _, size := range []int{100, 250, 500, 1000, 5000} {
+		b.Run(fmt.Sprintf("%dCAs", size), func(b *testing.B) {
+			l := test.NewLogger()
+			dir := b.TempDir()
+
+			ca, caKey, caBundle := buildCABundle(b, size)
+			caPath, certPath, keyPath := writePKIFiles(b, dir, ca, caKey, caBundle)
+
+			configBody := fmt.Sprintf(`pki:
+  ca: %s
+  cert: %s
+  key: %s
+`, caPath, certPath, keyPath)
+
+			configPath := filepath.Join(dir, "config.yml")
+			require.NoError(b, os.WriteFile(configPath, []byte(configBody), 0o600))
+
+			c := config.NewC(l)
+			require.NoError(b, c.Load(dir))
+
+			_, err := NewPKIFromConfig(l, c)
+			require.NoError(b, err)
+
+			b.ReportAllocs()
+			b.ResetTimer()
+
+			for b.Loop() {
+				c.ReloadConfig()
+			}
+		})
+	}
+}
+
+func buildCABundle(b *testing.B, count int) (cert.Certificate, []byte, []byte) {
+	b.Helper()
+	require.GreaterOrEqual(b, count, 1)
+
+	before := time.Now().Add(-24 * time.Hour)
+	after := time.Now().Add(24 * time.Hour)
+
+	ca, _, caKey, pem := cert_test.NewTestCaCert(
+		cert.Version2,
+		cert.Curve_CURVE25519,
+		before,
+		after,
+		nil,
+		nil,
+		nil,
+	)
+
+	buf := bytes.NewBuffer(pem)
+	buf.Write([]byte("\n# a comment!\n"))
+
+	for i := 1; i < count; i++ {
+		_, _, _, extraPEM := cert_test.NewTestCaCert(
+			cert.Version2,
+			cert.Curve_CURVE25519,
+			time.Now(),
+			time.Now().Add(time.Hour),
+			nil,
+			nil,
+			nil,
+		)
+		buf.Write([]byte("\n# a comment!\n"))
+		buf.Write(extraPEM)
+	}
+
+	return ca, caKey, buf.Bytes()
+}
+
+func writePKIFiles(b *testing.B, dir string, ca cert.Certificate, caKey []byte, caBundle []byte) (string, string, string) {
+	b.Helper()
+
+	networks := []netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")}
+
+	_, _, keyPEM, certPEM := cert_test.NewTestCert(
+		cert.Version2,
+		cert.Curve_CURVE25519,
+		ca,
+		caKey,
+		"reload-benchmark",
+		time.Now(),
+		time.Now().Add(time.Hour),
+		networks,
+		nil,
+		nil,
+	)
+
+	caPath := filepath.Join(dir, "ca.pem")
+	certPath := filepath.Join(dir, "cert.pem")
+	keyPath := filepath.Join(dir, "key.pem")
+
+	require.NoError(b, os.WriteFile(caPath, caBundle, 0o600))
+	require.NoError(b, os.WriteFile(certPath, certPEM, 0o600))
+	require.NoError(b, os.WriteFile(keyPath, keyPEM, 0o600))
+
+	return caPath, certPath, keyPath
+}

service/service.go

@@ -44,7 +44,10 @@ type Service struct {
 }
 
 func New(control *nebula.Control) (*Service, error) {
-	control.Start()
+	wait, err := control.Start()
+	if err != nil {
+		return nil, err
+	}
 
 	ctx := control.Context()
 	eg, ctx := errgroup.WithContext(ctx)
@@ -141,6 +144,12 @@ func New(control *nebula.Control) (*Service, error) {
 		}
 	})
 
+	// Add the nebula wait function to the group
+	eg.Go(func() error {
+		wait()
+		return nil
+	})
+
 	return &s, nil
 }
 

ssh.go

@@ -10,6 +10,7 @@ import (
 	"net"
 	"net/netip"
 	"os"
+	"path/filepath"
 	"reflect"
 	"runtime"
 	"runtime/pprof"
@@ -188,6 +189,12 @@ func configSSH(l *logrus.Logger, ssh *sshd.SSHServer, c *config.C) (func(), erro
 }
 
 func attachCommands(l *logrus.Logger, c *config.C, ssh *sshd.SSHServer, f *Interface) {
+	// sandboxDir defaults to a dir in temp. The intention is that end user will
+	// create this dir as needed. Overriding this config value to "" allows
+	// writing to anywhere in the system.
+	defaultDir := filepath.Join(os.TempDir(), "nebula-debug")
+	sandboxDir := c.GetString("sshd.sandbox_dir", defaultDir)
+
 	ssh.RegisterCommand(&sshd.Command{
 		Name:             "list-hostmap",
 		ShortDescription: "List all known previously connected hosts",
@@ -246,7 +253,9 @@ func attachCommands(l *logrus.Logger, c *config.C, ssh *sshd.SSHServer, f *Inter
 	ssh.RegisterCommand(&sshd.Command{
 		Name:             "start-cpu-profile",
 		ShortDescription: "Starts a cpu profile and write output to the provided file, ex: `cpu-profile.pb.gz`",
-		Callback:         sshStartCpuProfile,
+		Callback: func(fs any, a []string, w sshd.StringWriter) error {
+			return sshStartCpuProfile(sandboxDir, fs, a, w)
+		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -261,7 +270,9 @@ func attachCommands(l *logrus.Logger, c *config.C, ssh *sshd.SSHServer, f *Inter
 	ssh.RegisterCommand(&sshd.Command{
 		Name:             "save-heap-profile",
 		ShortDescription: "Saves a heap profile to the provided path, ex: `heap-profile.pb.gz`",
-		Callback:         sshGetHeapProfile,
+		Callback: func(fs any, a []string, w sshd.StringWriter) error {
+			return sshGetHeapProfile(sandboxDir, fs, a, w)
+		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -273,7 +284,9 @@ func attachCommands(l *logrus.Logger, c *config.C, ssh *sshd.SSHServer, f *Inter
 	ssh.RegisterCommand(&sshd.Command{
 		Name:             "save-mutex-profile",
 		ShortDescription: "Saves a mutex profile to the provided path, ex: `mutex-profile.pb.gz`",
-		Callback:         sshGetMutexProfile,
+		Callback: func(fs any, a []string, w sshd.StringWriter) error {
+			return sshGetMutexProfile(sandboxDir, fs, a, w)
+		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -506,13 +519,43 @@ func sshListLighthouseMap(lightHouse *LightHouse, a any, w sshd.StringWriter) er
 	return nil
 }
 
-func sshStartCpuProfile(fs any, a []string, w sshd.StringWriter) error {
+// sshSanitizeFilePath validates that the given file path is within the sandbox directory.
+// If sandboxDir is empty, the path is returned as-is for backwards compatibility.
+func sshSanitizeFilePath(sandboxDir, filePath string) (string, error) {
+	if sandboxDir == "" {
+		return filePath, nil
+	}
+
+	// Clean and resolve the path relative to the sandbox directory
+	if !filepath.IsAbs(filePath) {
+		filePath = filepath.Join(sandboxDir, filePath)
+	}
+	cleaned := filepath.Clean(filePath)
+
+	// Ensure the resolved path is within the sandbox directory
+	cleanedSandbox := filepath.Clean(sandboxDir)
+	if cleaned == cleanedSandbox {
+		return "", fmt.Errorf("path %q resolves to the sandbox directory itself %q", filePath, sandboxDir)
+	}
+	if !strings.HasPrefix(cleaned, cleanedSandbox+string(filepath.Separator)) {
+		return "", fmt.Errorf("path %q is outside the sandbox directory %q", filePath, sandboxDir)
+	}
+
+	return cleaned, nil
+}
+
+func sshStartCpuProfile(sandboxDir string, fs any, a []string, w sshd.StringWriter) error {
 	if len(a) == 0 {
 		err := w.WriteLine("No path to write profile provided")
 		return err
 	}
 
-	file, err := os.Create(a[0])
+	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	if err != nil {
+		return w.WriteLine(err.Error())
+	}
+
+	file, err := os.Create(filePath)
 	if err != nil {
 		err = w.WriteLine(fmt.Sprintf("Unable to create profile file: %s", err))
 		return err
@@ -676,12 +719,17 @@ func sshChangeRemote(ifce *Interface, fs any, a []string, w sshd.StringWriter) e
 	return w.WriteLine("Changed")
 }
 
-func sshGetHeapProfile(fs any, a []string, w sshd.StringWriter) error {
+func sshGetHeapProfile(sandboxDir string, fs any, a []string, w sshd.StringWriter) error {
 	if len(a) == 0 {
 		return w.WriteLine("No path to write profile provided")
 	}
 
-	file, err := os.Create(a[0])
+	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	if err != nil {
+		return w.WriteLine(err.Error())
+	}
+
+	file, err := os.Create(filePath)
 	if err != nil {
 		err = w.WriteLine(fmt.Sprintf("Unable to create profile file: %s", err))
 		return err
@@ -712,12 +760,17 @@ func sshMutexProfileFraction(fs any, a []string, w sshd.StringWriter) error {
 	return w.WriteLine(fmt.Sprintf("New value: %d. Old value: %d", newRate, oldRate))
 }
 
-func sshGetMutexProfile(fs any, a []string, w sshd.StringWriter) error {
+func sshGetMutexProfile(sandboxDir string, fs any, a []string, w sshd.StringWriter) error {
 	if len(a) == 0 {
 		return w.WriteLine("No path to write profile provided")
 	}
 
-	file, err := os.Create(a[0])
+	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	if err != nil {
+		return w.WriteLine(err.Error())
+	}
+
+	file, err := os.Create(filePath)
 	if err != nil {
 		return w.WriteLine(fmt.Sprintf("Unable to create profile file: %s", err))
 	}

udp/conn.go

@@ -16,7 +16,7 @@ type EncReader func(
 type Conn interface {
 	Rebind() error
 	LocalAddr() (netip.AddrPort, error)
-	ListenOut(r EncReader)
+	ListenOut(r EncReader) error
 	WriteTo(b []byte, addr netip.AddrPort) error
 	ReloadConfig(c *config.C)
 	SupportsMultipleReaders() bool
@@ -31,8 +31,8 @@ func (NoopConn) Rebind() error {
 func (NoopConn) LocalAddr() (netip.AddrPort, error) {
 	return netip.AddrPort{}, nil
 }
-func (NoopConn) ListenOut(_ EncReader) {
-	return
+func (NoopConn) ListenOut(_ EncReader) error {
+	return nil
 }
 func (NoopConn) SupportsMultipleReaders() bool {
 	return false

udp/udp_darwin.go

@@ -165,7 +165,7 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
 	return func() {}
 }
 
-func (u *StdConn) ListenOut(r EncReader) {
+func (u *StdConn) ListenOut(r EncReader) error {
 	buffer := make([]byte, MTU)
 
 	for {
@@ -173,8 +173,7 @@ func (u *StdConn) ListenOut(r EncReader) {
 		n, rua, err := u.ReadFromUDPAddrPort(buffer)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-				return
+				return err
 			}
 
 			u.l.WithError(err).Error("unexpected udp socket receive error")

udp/udp_generic.go

@@ -10,11 +10,9 @@ package udp
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"net/netip"
-	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
@@ -73,25 +71,14 @@ type rawMessage struct {
 	Len uint32
 }
 
-func (u *GenericConn) ListenOut(r EncReader) {
+func (u *GenericConn) ListenOut(r EncReader) error {
 	buffer := make([]byte, MTU)
 
-	var lastRecvErr time.Time
-
 	for {
 		// Just read one packet at a time
 		n, rua, err := u.ReadFromUDPAddrPort(buffer)
 		if err != nil {
-			if errors.Is(err, net.ErrClosed) {
-				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-				return
-			}
-			// Dampen unexpected message warns to once per minute
-			if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
-				lastRecvErr = time.Now()
-				u.l.WithError(err).Warn("unexpected udp socket receive error")
-			}
-			continue
+			return err
 		}
 
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])

udp/udp_linux.go

@@ -21,7 +21,6 @@ import (
 type StdConn struct {
 	udpConn *net.UDPConn
 	rawConn syscall.RawConn
-	sockFd  int
 	isV4    bool
 	l       *logrus.Logger
 	batch   int
@@ -45,12 +44,8 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
 	if multi {
 		lc.Control = setReusePort
 	}
-	network := "udp"
-	if ip.Is4() { //todo I am not a fan of this, would prefer a default where we always bind dual-stack unless given a non-wildcard v4 addr
-		network = "udp4"
-	}
 	//this context is only used during the bind operation, you can't cancel it to kill the socket
-	pc, err := lc.ListenPacket(context.Background(), network, listen.String())
+	pc, err := lc.ListenPacket(context.Background(), "udp", listen.String())
 	if err != nil {
 		return nil, fmt.Errorf("unable to open socket: %s", err)
 	}
@@ -60,25 +55,22 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
 		_ = udpConn.Close()
 		return nil, err
 	}
-
-	//steal the socket's fd for sending
-	var sockFd uintptr
-	err = rawConn.Control(func(fd uintptr) {
-		sockFd = fd
-	})
-	if err != nil {
-		_ = udpConn.Close()
-		return nil, err
-	}
-
-	return &StdConn{
+	//gotta find out if we got an AF_INET6 socket or not:
+	out := &StdConn{
 		udpConn: udpConn,
 		rawConn: rawConn,
-		sockFd:  int(sockFd),
-		isV4:    ip.Is4(),
 		l:       l,
 		batch:   batch,
-	}, err
+	}
+
+	af, err := out.getSockOptInt(unix.SO_DOMAIN)
+	if err != nil {
+		_ = out.Close()
+		return nil, err
+	}
+	out.isV4 = af == unix.AF_INET
+
+	return out, nil
 }
 
 func (u *StdConn) SupportsMultipleReaders() bool {
@@ -143,8 +135,19 @@ func (u *StdConn) GetSoMark() (int, error) {
 }
 
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
-	addr := u.udpConn.LocalAddr()
-	return netip.ParseAddrPort(addr.String())
+	a := u.udpConn.LocalAddr()
+
+	switch v := a.(type) {
+	case *net.UDPAddr:
+		addr, ok := netip.AddrFromSlice(v.IP)
+		if !ok {
+			return netip.AddrPort{}, fmt.Errorf("LocalAddr returned invalid IP address: %s", v.IP)
+		}
+		return netip.AddrPortFrom(addr, uint16(v.Port)), nil
+
+	default:
+		return netip.AddrPort{}, fmt.Errorf("LocalAddr returned: %#v", a)
+	}
 }
 
 func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
@@ -168,7 +171,7 @@ func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
 	return int(n), true, nil
 }
 
-func (u *StdConn) listenOutSingle(r EncReader) {
+func (u *StdConn) listenOutSingle(r EncReader) error {
 	var err error
 	var n int
 	var from netip.AddrPort
@@ -177,15 +180,14 @@ func (u *StdConn) listenOutSingle(r EncReader) {
 	for {
 		n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
 		if err != nil {
-			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-			return
+			return err
 		}
 		from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
 		r(from, buffer[:n])
 	}
 }
 
-func (u *StdConn) listenOutBatch(r EncReader) {
+func (u *StdConn) listenOutBatch(r EncReader) error {
 	var ip netip.Addr
 	var n int
 	var operr error
@@ -202,12 +204,10 @@ func (u *StdConn) listenOutBatch(r EncReader) {
 	for {
 		err := u.rawConn.Read(reader)
 		if err != nil {
-			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-			return
+			return err
 		}
 		if operr != nil {
-			u.l.WithError(operr).Debug("operr: udp socket is closed, exiting read loop")
-			return
+			return operr
 		}
 
 		for i := 0; i < n; i++ {
@@ -222,76 +222,17 @@ func (u *StdConn) listenOutBatch(r EncReader) {
 	}
 }
 
-func (u *StdConn) ListenOut(r EncReader) {
+func (u *StdConn) ListenOut(r EncReader) error {
 	if u.batch == 1 {
-		//save some ram by not calling PrepareRawMessages for fields we won't use
-		//we could also make this path more common by calling recvmmsg with msgs[:1],
-		//but that's still the recvmmsg syscall, which would be a change
-		u.listenOutSingle(r)
+		return u.listenOutSingle(r)
 	} else {
-		u.listenOutBatch(r)
+		return u.listenOutBatch(r)
 	}
 }
 
 func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
-	if u.isV4 {
-		return u.writeTo4(b, ip)
-	}
-	return u.writeTo6(b, ip)
-}
-
-func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
-	var rsa unix.RawSockaddrInet6
-	rsa.Family = unix.AF_INET6
-	rsa.Addr = ip.Addr().As16()
-	binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
-
-	for {
-		_, _, err := unix.Syscall6(
-			unix.SYS_SENDTO,
-			uintptr(u.sockFd),
-			uintptr(unsafe.Pointer(&b[0])),
-			uintptr(len(b)),
-			uintptr(0),
-			uintptr(unsafe.Pointer(&rsa)),
-			uintptr(unix.SizeofSockaddrInet6),
-		)
-
-		if err != 0 {
-			return &net.OpError{Op: "sendto", Err: err}
-		}
-
-		return nil
-	}
-}
-
-func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
-	if !ip.Addr().Is4() {
-		return ErrInvalidIPv6RemoteForSocket
-	}
-
-	var rsa unix.RawSockaddrInet4
-	rsa.Family = unix.AF_INET
-	rsa.Addr = ip.Addr().As4()
-	binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
-
-	for {
-		_, _, err := unix.Syscall6(
-			unix.SYS_SENDTO,
-			uintptr(u.sockFd),
-			uintptr(unsafe.Pointer(&b[0])),
-			uintptr(len(b)),
-			uintptr(0),
-			uintptr(unsafe.Pointer(&rsa)),
-			uintptr(unix.SizeofSockaddrInet4),
-		)
-
-		if err != 0 {
-			return &net.OpError{Op: "sendto", Err: err}
-		}
-
-		return nil
-	}
+	_, err := u.udpConn.WriteToUDPAddrPort(b, ip)
+	return err
 }
 
 func (u *StdConn) ReloadConfig(c *config.C) {

udp/udp_rio_windows.go

@@ -140,7 +140,7 @@ func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
 	return nil
 }
 
-func (u *RIOConn) ListenOut(r EncReader) {
+func (u *RIOConn) ListenOut(r EncReader) error {
 	buffer := make([]byte, MTU)
 
 	var lastRecvErr time.Time
@@ -151,8 +151,7 @@ func (u *RIOConn) ListenOut(r EncReader) {
 
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
-				return
+				return err
 			}
 			// Dampen unexpected message warns to once per minute
 			if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {

udp/udp_tester.go

@@ -6,6 +6,7 @@ package udp
 import (
 	"io"
 	"net/netip"
+	"os"
 	"sync/atomic"
 
 	"github.com/sirupsen/logrus"
@@ -106,11 +107,11 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return nil
 }
 
-func (u *TesterConn) ListenOut(r EncReader) {
+func (u *TesterConn) ListenOut(r EncReader) error {
 	for {
 		p, ok := <-u.RxPackets
 		if !ok {
-			return
+			return os.ErrClosed
 		}
 		r(p.From, p.Data)
 	}