drop runtime.LockOSThread() on listenOut

.github/workflows/release.yml

@@ -209,11 +209,10 @@ 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}" \
+            --title "Release ${{ github.ref_name }}" \
-            "${GITHUB_REF_NAME}" \
+            "${{ github.ref_name }}" \
             SHASUM256.txt *-latest/*.zip *-latest/*.tar.gz

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

@@ -29,17 +29,6 @@ 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,7 +124,6 @@ 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 ".", "/vagrant", disabled: true
+  config.vm.synced_folder "../build", "/nebula", type: "rsync"
 end

CODEOWNERS

@@ -1 +0,0 @@
-#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 ([source code](https://github.com/DefinedNet/mobile_nebula))
+#### Mobile
 
 - [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,8 +76,6 @@ 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,14 +1,11 @@
 package cert
 
 import (
-	"bufio"
-	"bytes"
-	"encoding/pem"
 	"errors"
 	"fmt"
-	"io"
 	"net/netip"
 	"slices"
+	"strings"
 	"time"
 )
 
@@ -32,46 +29,22 @@ 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 {
-	scanner := bufio.NewScanner(r)
+		caPEMs, err = pool.AddCAFromPEM(caPEMs)
-	scanner.Split(SplitPEM)
+		if errors.Is(err, ErrExpired) {
-
+			expired = true
-	for scanner.Scan() {
+			err = nil
-		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)) == "" {
-		err = pool.AddCA(c)
+			break
-		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,10 +1,7 @@
 package cert
 
 import (
-	"bytes"
-	"io"
 	"net/netip"
-	"strings"
 	"testing"
 	"time"
 
@@ -115,60 +112,6 @@ 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,12 +44,7 @@ func swap(r, s []byte) ([]byte, []byte, error) {
 	}
 	sNormalized := nMod.Nat().Sub(bigS, nMod)
 
-	result := sNormalized.Bytes(nMod)
+	return r, sNormalized.Bytes(nMod), nil
-	for len(result) > 1 && result[0] == 0 {
-		result = result[1:]
-	}
-
-	return r, result, nil
 }
 
 func Normalize(sig []byte) ([]byte, error) {

cert/pem.go

@@ -1,66 +1,12 @@
 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"
@@ -91,7 +37,19 @@ func UnmarshalCertificateFromPEM(b []byte) (Certificate, []byte, error) {
 		return nil, r, ErrInvalidPEMBlock
 	}
 
-	c, err := unmarshalCertificateBlock(p)
+	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
+	}
+
 	if err != nil {
 		return nil, r, err
 	}
@@ -100,20 +58,6 @@ 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,88 +1,12 @@
 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,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/slackhq/nebula/cert"
@@ -39,15 +40,21 @@ func verify(args []string, out io.Writer, errOut io.Writer) error {
 		return err
 	}
 
-	caFile, err := os.Open(*vf.caPath)
+	rawCACert, err := os.ReadFile(*vf.caPath)
 	if err != nil {
 		return fmt.Errorf("error while reading ca: %w", err)
 	}
-	defer caFile.Close()
 
-	caPool, err := cert.NewCAPoolFromPEMReader(caFile)
+	caPool := cert.NewCAPool()
-	if err != nil && !errors.Is(err, cert.ErrExpired) {
+	for {
-		return fmt.Errorf("error while adding ca cert to pool: %w", err)
+		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
+		}
 	}
 
 	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.ErrorIs(t, err, cert.ErrInvalidPEMBlock)
+	require.EqualError(t, err, "error while adding ca cert to pool: input did not contain a valid PEM encoded block")
 
 	// make a ca for later
 	caPub, caPriv, _ := ed25519.GenerateKey(rand.Reader)

cmd/nebula-service/main.go

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

control.go

@@ -2,11 +2,9 @@ package nebula
 
 import (
 	"context"
-	"errors"
 	"net/netip"
 	"os"
 	"os/signal"
-	"sync"
 	"syscall"
 
 	"github.com/sirupsen/logrus"
@@ -15,16 +13,6 @@ 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
 
@@ -38,9 +26,6 @@ type controlHostLister interface {
 }
 
 type Control struct {
-	stateLock sync.Mutex
-	state     RunState
-
 	f                      *Interface
 	l                      *logrus.Logger
 	ctx                    context.Context
@@ -64,21 +49,10 @@ type ControlHostInfo struct {
 	CurrentRelaysThroughMe []netip.Addr     `json:"currentRelaysThroughMe"`
 }
 
-// Start actually runs nebula, this is a nonblocking call.
+// Start actually runs nebula, this is a nonblocking call. To block use Control.ShutdownBlock()
-// The returned function can be used to wait for nebula to fully stop.
+func (c *Control) Start() {
-func (c *Control) Start() (func(), error) {
-	c.stateLock.Lock()
-	if c.state != Stopped {
-		c.stateLock.Unlock()
-		return nil, ErrAlreadyStarted
-	}
-
 	// Activate the interface
-	err := c.f.activate()
+	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 {
@@ -98,33 +72,15 @@ func (c *Control) Start() (func(), error) {
 	}
 
 	// Start reading packets.
-	c.state = Started
+	c.f.run()
-	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 is a non-blocking call that signals nebula to close all tunnels and shut down
+// Stop signals nebula to shutdown and close all tunnels, returns after the shutdown is complete
 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()
@@ -133,9 +89,7 @@ func (c *Control) Stop() {
 	if err := c.f.Close(); err != nil {
 		c.l.WithError(err).Error("Close interface failed")
 	}
-	c.stateLock.Lock()
+	c.l.Info("Goodbye")
-	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,24 +84,30 @@ 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 all existing registrations
+    -- Remove our old dissector
     DissectorTable.get("udp.port"):remove_all(nebula)
 
-    if nebula.prefs.all_ports then
+    if nebula.prefs.all_ports and default_settings.all_ports ~= nebula.prefs.all_ports then
-        -- Register on every port for hole punch capture
+        default_settings.all_port = nebula.prefs.all_ports
+
         for i=0, 65535 do
             DissectorTable.get("udp.port"):add(i, nebula)
         end
-    else
+
-        -- Register on the configured port only
+        -- no need to establish again on specific ports
-        DissectorTable.get("udp.port"):add(nebula.prefs.port, nebula)
+        return
     end
 
-    default_settings.all_ports = nebula.prefs.all_ports
+
-    default_settings.port = nebula.prefs.port
+    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
 end
 
 DissectorTable.get("udp.port"):add(default_settings.port, nebula)

e2e/handshake_manager_test.go

@@ -1,565 +0,0 @@
-//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,8 +12,6 @@ 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"
 )
@@ -367,106 +365,3 @@ 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,12 +204,6 @@ 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", outboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
+		l.WithField("action", inboundAction).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.72
+	github.com/miekg/dns v1.1.70
-	github.com/miekg/pkcs11 v1.1.2
+	github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b
 	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.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
+github.com/miekg/dns v1.1.70 h1:DZ4u2AV35VJxdD9Fo9fIWm119BsQL5cZU1cQ9s0LkqA=
-github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
+github.com/miekg/dns v1.1.70/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
-github.com/miekg/pkcs11 v1.1.2 h1:/VxmeAX5qU6Q3EwafypogwWbYryHFmF2RpkJmw3m4MQ=
+github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b h1:J/AzCvg5z0Hn1rqZUJjpbzALUmkKX0Zwbc/i4fw7Sfk=
-github.com/miekg/pkcs11 v1.1.2/go.mod h1:XsNlhZGX73bx86s2hdc/FuaLm2CPZJemRLMA+WTFxgs=
+github.com/miekg/pkcs11 v1.1.2-0.20231115102856-9078ad6b9d4b/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,8 @@ import (
 	"fmt"
 	"io"
 	"net/netip"
-	"sync"
+	"os"
+	"runtime"
 	"sync/atomic"
 	"time"
 
@@ -87,7 +88,6 @@ type Interface struct {
 
 	writers []udp.Conn
 	readers []io.ReadWriteCloser
-	wg      sync.WaitGroup
 
 	metricHandshakes    metrics.Histogram
 	messageMetrics      *MessageMetrics
@@ -210,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() error {
+func (f *Interface) activate() {
 	// actually turn on tun dev
 
 	addr, err := f.outside.LocalAddr()
@@ -238,36 +238,28 @@ func (f *Interface) activate() error {
 		if i > 0 {
 			reader, err = f.inside.NewMultiQueueReader()
 			if err != nil {
-				return err
+				f.l.Fatal(err)
 			}
 		}
 		f.readers[i] = reader
 	}
 
-	if err = f.inside.Activate(); err != nil {
+	if err := f.inside.Activate(); err != nil {
 		f.inside.Close()
-		return err
+		f.l.Fatal(err)
 	}
-
-	return nil
 }
 
-func (f *Interface) run() (func(), error) {
+func (f *Interface) run() {
 	// Launch n queues to read packets from udp
 	for i := 0; i < f.routines; i++ {
-		f.wg.Go(func() {
+		go f.listenOut(i)
-			f.listenOut(i)
-		})
 	}
 
 	// Launch n queues to read packets from tun dev
 	for i := 0; i < f.routines; i++ {
-		f.wg.Go(func() {
+		go f.listenIn(f.readers[i], i)
-			f.listenIn(f.readers[i], i)
-		})
 	}
-
-	return f.wg.Wait, nil
 }
 
 func (f *Interface) listenOut(i int) {
@@ -285,19 +277,14 @@ func (f *Interface) listenOut(i int) {
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
 
-	err := li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
+	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) {
+	runtime.LockOSThread()
+
 	packet := make([]byte, mtu)
 	out := make([]byte, mtu)
 	fwPacket := &firewall.Packet{}
@@ -308,17 +295,17 @@ func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
 	for {
 		n, err := reader.Read(packet)
 		if err != nil {
-			if !f.closed.Load() {
+			if errors.Is(err, os.ErrClosed) && f.closed.Load() {
-				f.l.WithError(err).WithField("reader", i).Error("Error while reading outbound packet, closing")
+				return
-				//TODO: Trigger Control to close
 			}
-			break
+
+			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)
 		}
 
 		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) {
@@ -493,26 +480,23 @@ func (f *Interface) GetCertState() *CertState {
 }
 
 func (f *Interface) Close() error {
-	var err error
 	f.closed.Store(true)
 
-	// Release the udp readers
+	for _, u := range f.writers {
-	for i, u := range f.writers {
+		err := u.Close()
-		err = u.Close()
 		if err != nil {
-			f.l.WithError(err).WithField("writer", i).Error("Error while closing udp socket")
+			f.l.WithError(err).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
+			continue // f.readers[0] is f.inside, which we want to save for last
 		}
-		if err = r.Close(); err != nil {
+		if err := r.Close(); err != nil {
-			f.l.WithError(err).WithField("reader", i).Error("Error while closing tun reader")
+			f.l.WithError(err).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{
-		f:                      ifce,
+		ifce,
-		l:                      l,
+		l,
-		ctx:                    ctx,
+		ctx,
-		cancel:                 cancel,
+		cancel,
-		sshStart:               sshStart,
+		sshStart,
-		statsStart:             statsStart,
+		statsStart,
-		dnsStart:               dnsStart,
+		dnsStart,
-		lighthouseStart:        lightHouse.StartUpdateWorker,
+		lightHouse.StartUpdateWorker,
-		connectionManagerStart: connManager.Start,
+		connManager.Start,
 	}, nil
 }
 

outside.go

@@ -190,13 +190,6 @@ 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,6 +26,7 @@ import (
 
 type tun struct {
 	io.ReadWriteCloser
+	fd          int
 	Device      string
 	vpnNetworks []netip.Prefix
 	MaxMTU      int
@@ -84,7 +85,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|unix.O_NONBLOCK, 0)
+	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
 	if err != nil {
 		// If /dev/net/tun doesn't exist, try to create it (will happen in docker)
 		if os.IsNotExist(err) {
@@ -97,7 +98,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|unix.O_NONBLOCK, 0)
+			fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
 			if err != nil {
 				return nil, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
 			}
@@ -135,6 +136,7 @@ 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),
@@ -259,6 +261,29 @@ func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
 	return r
 }
 
+func (t *tun) Write(b []byte) (int, error) {
+	var nn int
+	maximum := len(b)
+
+	for {
+		n, err := unix.Write(t.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) deviceBytes() (o [16]byte) {
 	for i, c := range t.Device {
 		o[i] = byte(c)
@@ -684,23 +709,15 @@ func (t *tun) updateRoutes(r netlink.RouteUpdate) {
 func (t *tun) Close() error {
 	if t.routeChan != nil {
 		close(t.routeChan)
-		t.routeChan = nil
-	}
-
-	if t.ioctlFd > 0 {
-		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()
+		_ = t.ReadWriteCloser.Close()
-		if err != nil {
+	}
-			t.l.WithField("error", err).Error("Failed to close tun file")
+
-			return err
+	if t.ioctlFd > 0 {
-		}
+		_ = os.NewFile(t.ioctlFd, "ioctlFd").Close()
+		t.ioctlFd = 0
 	}
 
 	return nil

pki.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"net/netip"
 	"os"
@@ -488,25 +487,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")
 	}
 
-	var caReader io.ReadCloser
-	var err error
-
 	if strings.Contains(caPathOrPEM, "-----BEGIN") {
-		caReader = io.NopCloser(strings.NewReader(caPathOrPEM))
+		rawCA = []byte(caPathOrPEM)
+
 	} else {
-		caReader, err = os.Open(caPathOrPEM)
+		rawCA, err = os.ReadFile(caPathOrPEM)
 		if err != nil {
 			return nil, fmt.Errorf("unable to read pki.ca file %s: %s", caPathOrPEM, err)
 		}
 	}
-	defer caReader.Close()
 
-	caPool, err := cert.NewCAPoolFromPEMReader(caReader)
+	caPool, err := cert.NewCAPoolFromPEM(rawCA)
 	if errors.Is(err, cert.ErrExpired) {
 		var expired int
 		for _, crt := range caPool.CAs {

pki_hup_benchmark_test.go

@@ -1,121 +0,0 @@
-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,10 +44,7 @@ type Service struct {
 }
 
 func New(control *nebula.Control) (*Service, error) {
-	wait, err := control.Start()
+	control.Start()
-	if err != nil {
-		return nil, err
-	}
 
 	ctx := control.Context()
 	eg, ctx := errgroup.WithContext(ctx)
@@ -144,12 +141,6 @@ 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,7 +10,6 @@ import (
 	"net"
 	"net/netip"
 	"os"
-	"path/filepath"
 	"reflect"
 	"runtime"
 	"runtime/pprof"
@@ -189,12 +188,6 @@ 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",
@@ -253,9 +246,7 @@ 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: func(fs any, a []string, w sshd.StringWriter) error {
+		Callback:         sshStartCpuProfile,
-			return sshStartCpuProfile(sandboxDir, fs, a, w)
-		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -270,9 +261,7 @@ 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: func(fs any, a []string, w sshd.StringWriter) error {
+		Callback:         sshGetHeapProfile,
-			return sshGetHeapProfile(sandboxDir, fs, a, w)
-		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -284,9 +273,7 @@ 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: func(fs any, a []string, w sshd.StringWriter) error {
+		Callback:         sshGetMutexProfile,
-			return sshGetMutexProfile(sandboxDir, fs, a, w)
-		},
 	})
 
 	ssh.RegisterCommand(&sshd.Command{
@@ -519,43 +506,13 @@ func sshListLighthouseMap(lightHouse *LightHouse, a any, w sshd.StringWriter) er
 	return nil
 }
 
-// sshSanitizeFilePath validates that the given file path is within the sandbox directory.
+func sshStartCpuProfile(fs any, a []string, w sshd.StringWriter) error {
-// 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
 	}
 
-	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	file, err := os.Create(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
@@ -719,17 +676,12 @@ func sshChangeRemote(ifce *Interface, fs any, a []string, w sshd.StringWriter) e
 	return w.WriteLine("Changed")
 }
 
-func sshGetHeapProfile(sandboxDir string, fs any, a []string, w sshd.StringWriter) error {
+func sshGetHeapProfile(fs any, a []string, w sshd.StringWriter) error {
 	if len(a) == 0 {
 		return w.WriteLine("No path to write profile provided")
 	}
 
-	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	file, err := os.Create(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
@@ -760,17 +712,12 @@ 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(sandboxDir string, fs any, a []string, w sshd.StringWriter) error {
+func sshGetMutexProfile(fs any, a []string, w sshd.StringWriter) error {
 	if len(a) == 0 {
 		return w.WriteLine("No path to write profile provided")
 	}
 
-	filePath, err := sshSanitizeFilePath(sandboxDir, a[0])
+	file, err := os.Create(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) error
+	ListenOut(r EncReader)
 	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) error {
+func (NoopConn) ListenOut(_ EncReader) {
-	return nil
+	return
 }
 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) error {
+func (u *StdConn) ListenOut(r EncReader) {
 	buffer := make([]byte, MTU)
 
 	for {
@@ -173,7 +173,8 @@ func (u *StdConn) ListenOut(r EncReader) error {
 		n, rua, err := u.ReadFromUDPAddrPort(buffer)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				return err
+				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+				return
 			}
 
 			u.l.WithError(err).Error("unexpected udp socket receive error")

udp/udp_generic.go

@@ -10,9 +10,11 @@ package udp
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
@@ -71,14 +73,25 @@ type rawMessage struct {
 	Len uint32
 }
 
-func (u *GenericConn) ListenOut(r EncReader) error {
+func (u *GenericConn) ListenOut(r EncReader) {
 	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 {
-			return err
+			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
 		}
 
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])

udp/udp_linux.go

@@ -21,6 +21,7 @@ import (
 type StdConn struct {
 	udpConn *net.UDPConn
 	rawConn syscall.RawConn
+	sockFd  int
 	isV4    bool
 	l       *logrus.Logger
 	batch   int
@@ -44,8 +45,12 @@ 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(), "udp", listen.String())
+	pc, err := lc.ListenPacket(context.Background(), network, listen.String())
 	if err != nil {
 		return nil, fmt.Errorf("unable to open socket: %s", err)
 	}
@@ -55,22 +60,25 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
 		_ = udpConn.Close()
 		return nil, err
 	}
-	//gotta find out if we got an AF_INET6 socket or not:
-	out := &StdConn{
-		udpConn: udpConn,
-		rawConn: rawConn,
-		l:       l,
-		batch:   batch,
-	}
 
-	af, err := out.getSockOptInt(unix.SO_DOMAIN)
+	//steal the socket's fd for sending
+	var sockFd uintptr
+	err = rawConn.Control(func(fd uintptr) {
+		sockFd = fd
+	})
 	if err != nil {
-		_ = out.Close()
+		_ = udpConn.Close()
 		return nil, err
 	}
-	out.isV4 = af == unix.AF_INET
 
-	return out, nil
+	return &StdConn{
+		udpConn: udpConn,
+		rawConn: rawConn,
+		sockFd:  int(sockFd),
+		isV4:    ip.Is4(),
+		l:       l,
+		batch:   batch,
+	}, err
 }
 
 func (u *StdConn) SupportsMultipleReaders() bool {
@@ -135,19 +143,8 @@ func (u *StdConn) GetSoMark() (int, error) {
 }
 
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
-	a := u.udpConn.LocalAddr()
+	addr := u.udpConn.LocalAddr()
-
+	return netip.ParseAddrPort(addr.String())
-	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) {
@@ -171,7 +168,7 @@ func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
 	return int(n), true, nil
 }
 
-func (u *StdConn) listenOutSingle(r EncReader) error {
+func (u *StdConn) listenOutSingle(r EncReader) {
 	var err error
 	var n int
 	var from netip.AddrPort
@@ -180,14 +177,15 @@ func (u *StdConn) listenOutSingle(r EncReader) error {
 	for {
 		n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
 		if err != nil {
-			return err
+			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+			return
 		}
 		from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
 		r(from, buffer[:n])
 	}
 }
 
-func (u *StdConn) listenOutBatch(r EncReader) error {
+func (u *StdConn) listenOutBatch(r EncReader) {
 	var ip netip.Addr
 	var n int
 	var operr error
@@ -204,10 +202,12 @@ func (u *StdConn) listenOutBatch(r EncReader) error {
 	for {
 		err := u.rawConn.Read(reader)
 		if err != nil {
-			return err
+			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+			return
 		}
 		if operr != nil {
-			return operr
+			u.l.WithError(operr).Debug("operr: udp socket is closed, exiting read loop")
+			return
 		}
 
 		for i := 0; i < n; i++ {
@@ -222,17 +222,76 @@ func (u *StdConn) listenOutBatch(r EncReader) error {
 	}
 }
 
-func (u *StdConn) ListenOut(r EncReader) error {
+func (u *StdConn) ListenOut(r EncReader) {
 	if u.batch == 1 {
-		return u.listenOutSingle(r)
+		//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)
 	} else {
-		return u.listenOutBatch(r)
+		u.listenOutBatch(r)
 	}
 }
 
 func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
-	_, err := u.udpConn.WriteToUDPAddrPort(b, ip)
+	if u.isV4 {
-	return err
+		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
+	}
 }
 
 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) error {
+func (u *RIOConn) ListenOut(r EncReader) {
 	buffer := make([]byte, MTU)
 
 	var lastRecvErr time.Time
@@ -151,7 +151,8 @@ func (u *RIOConn) ListenOut(r EncReader) error {
 
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				return err
+				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 {

udp/udp_tester.go

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