Support multiple UDP source ports (multiport)

.github/workflows/smoke.yml

@@ -52,4 +52,12 @@ jobs:
       working-directory: ./.github/workflows/smoke
       run: NAME="smoke-p256" ./smoke.sh
 
+    - name: setup docker image for multiport
+      working-directory: ./.github/workflows/smoke
+      run: NAME="smoke-multiport" MULTIPORT_TX=true MULTIPORT_RX=true MULTIPORT_HANDSHAKE=true ./build.sh
+
+    - name: run smoke
+      working-directory: ./.github/workflows/smoke
+      run: NAME="smoke-multiport" ./smoke.sh
+
     timeout-minutes: 10

.github/workflows/smoke/genconfig.sh

@@ -48,6 +48,10 @@ listen:
 
 tun:
   dev: ${TUN_DEV:-tun0}
+  multiport:
+    tx_enabled: ${MULTIPORT_TX:-false}
+    rx_enabled: ${MULTIPORT_RX:-false}
+    tx_handshake: ${MULTIPORT_HANDSHAKE:-false}
 
 firewall:
   inbound_action: reject

Makefile

@@ -227,6 +227,10 @@ smoke-relay-docker: bin-docker
 	cd .github/workflows/smoke/ && ./build-relay.sh
 	cd .github/workflows/smoke/ && ./smoke-relay.sh
 
+smoke-multiport-docker: bin-docker
+	cd .github/workflows/smoke/ && NAME="smoke-multiport" MULTIPORT_TX=true MULTIPORT_RX=true MULTIPORT_HANDSHAKE=true ./build.sh
+	cd .github/workflows/smoke/ && NAME="smoke-multiport" ./smoke.sh
+
 smoke-docker-race: BUILD_ARGS = -race
 smoke-docker-race: CGO_ENABLED = 1
 smoke-docker-race: smoke-docker

examples/config.yml

@@ -284,6 +284,47 @@ tun:
   # SO_RCVBUFFORCE is used to avoid having to raise the system wide max
   #use_system_route_table_buffer_size: 0
 
+  # EXPERIMENTAL: This option may change or disappear in the future.
+  # Multiport spreads outgoing UDP packets across multiple UDP send ports,
+  # which allows nebula to work around any issues on the underlay network.
+  # Some example issues this could work around:
+  #  - UDP rate limits on a per flow basis.
+  #  - Partial underlay network failure in which some flows work and some don't
+  # Agreement is done during the handshake to decide if multiport mode will
+  # be used for a given tunnel (one side must have tx_enabled set, the other
+  # side must have rx_enabled set)
+  #
+  # NOTE: you cannot use multiport on a host if you are relying on UDP hole
+  # punching to get through a NAT or firewall.
+  #
+  # NOTE: Linux only (uses raw sockets to send). Also currently only works
+  # with IPv4 underlay network remotes.
+  #
+  # The default values are listed below:
+  #multiport:
+    # This host support sending via multiple UDP ports.
+    #tx_enabled: false
+    #
+    # This host supports receiving packets sent from multiple UDP ports.
+    #rx_enabled: false
+    #
+    # How many UDP ports to use when sending. The lowest source port will be
+    # listen.port and go up to (but not including) listen.port + tx_ports.
+    #tx_ports: 100
+    #
+    # NOTE: All of your hosts must be running a version of Nebula that supports
+    # multiport if you want to enable this feature. Older versions of Nebula
+    # will be confused by these multiport handshakes.
+    #
+    # If handshakes are not getting a response, attempt to transmit handshakes
+    # using random UDP source ports (to get around partial underlay network
+    # failures).
+    #tx_handshake: false
+    #
+    # How many unresponded handshakes we should send before we attempt to
+    # send multiport handshakes.
+    #tx_handshake_delay: 2
+
 # Configure logging level
 logging:
   # panic, fatal, error, warning, info, or debug. Default is info and is reloadable.

firewall/packet.go

@@ -3,6 +3,7 @@ package firewall
 import (
 	"encoding/json"
 	"fmt"
+	mathrand "math/rand"
 	"net/netip"
 )
 
@@ -60,3 +61,30 @@ func (fp Packet) MarshalJSON() ([]byte, error) {
 		"Fragment":   fp.Fragment,
 	})
 }
+
+// UDPSendPort calculates the UDP port to send from when using multiport mode.
+// The result will be from [0, numBuckets)
+func (fp Packet) UDPSendPort(numBuckets int) uint16 {
+	if numBuckets <= 1 {
+		return 0
+	}
+
+	// If there is no port (like an ICMP packet), pick a random UDP send port
+	if fp.LocalPort == 0 {
+		return uint16(mathrand.Intn(numBuckets))
+	}
+
+	// A decent enough 32bit hash function
+	// Prospecting for Hash Functions
+	// - https://nullprogram.com/blog/2018/07/31/
+	// - https://github.com/skeeto/hash-prospector
+	//   [16 21f0aaad 15 d35a2d97 15] = 0.10760229515479501
+	x := (uint32(fp.LocalPort) << 16) | uint32(fp.RemotePort)
+	x ^= x >> 16
+	x *= 0x21f0aaad
+	x ^= x >> 15
+	x *= 0xd35a2d97
+	x ^= x >> 15
+
+	return uint16(x) % uint16(numBuckets)
+}

handshake_ix.go

@@ -9,6 +9,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/header"
+	"github.com/slackhq/nebula/udp"
 )
 
 // NOISE IX Handshakes
@@ -74,6 +75,15 @@ func ixHandshakeStage0(f *Interface, hh *HandshakeHostInfo) bool {
 		},
 	}
 
+	if f.multiPort.Tx || f.multiPort.Rx {
+		hs.Details.InitiatorMultiPort = &MultiPortDetails{
+			RxSupported: f.multiPort.Rx,
+			TxSupported: f.multiPort.Tx,
+			BasePort:    uint32(f.multiPort.TxBasePort),
+			TotalPorts:  uint32(f.multiPort.TxPorts),
+		}
+	}
+
 	hsBytes, err := hs.Marshal()
 	if err != nil {
 		f.l.WithError(err).WithField("vpnAddrs", hh.hostinfo.vpnAddrs).
@@ -243,13 +253,35 @@ func ixHandshakeStage1(f *Interface, via ViaSender, packet []byte, h *header.H)
 		return
 	}
 
+	var multiportTx, multiportRx bool
+	if f.multiPort.Rx || f.multiPort.Tx {
+		if hs.Details.InitiatorMultiPort != nil {
+			multiportTx = hs.Details.InitiatorMultiPort.RxSupported && f.multiPort.Tx
+			multiportRx = hs.Details.InitiatorMultiPort.TxSupported && f.multiPort.Rx
+		}
+
+		hs.Details.ResponderMultiPort = &MultiPortDetails{
+			TxSupported: f.multiPort.Tx,
+			RxSupported: f.multiPort.Rx,
+			BasePort:    uint32(f.multiPort.TxBasePort),
+			TotalPorts:  uint32(f.multiPort.TxPorts),
+		}
+	}
+	if hs.Details.InitiatorMultiPort != nil && hs.Details.InitiatorMultiPort.BasePort != uint32(via.UdpAddr.Port()) {
+		// The other side sent us a handshake from a different port, make sure
+		// we send responses back to the BasePort
+		via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), uint16(hs.Details.InitiatorMultiPort.BasePort))
+	}
+
 	hostinfo := &HostInfo{
 		ConnectionState:   ci,
 		localIndexId:      myIndex,
 		remoteIndexId:     hs.Details.InitiatorIndex,
 		vpnAddrs:          vpnAddrs,
 		HandshakePacket:   make(map[uint8][]byte, 0),
 		lastHandshakeTime: hs.Details.Time,
+		multiportTx:       multiportTx,
+		multiportRx:       multiportRx,
 		relayState: RelayState{
 			relays:         nil,
 			relayForByAddr: map[netip.Addr]*Relay{},
@@ -267,6 +299,8 @@ func ixHandshakeStage1(f *Interface, via ViaSender, packet []byte, h *header.H)
 		"initiatorIndex": hs.Details.InitiatorIndex,
 		"responderIndex": hs.Details.ResponderIndex,
 		"remoteIndex":    h.RemoteIndex,
+		"multiportTx":    multiportTx,
+		"multiportRx":    multiportRx,
 		"handshake":      m{"stage": 1, "style": "ix_psk0"},
 	})
 
@@ -346,6 +380,10 @@ func ixHandshakeStage1(f *Interface, via ViaSender, packet []byte, h *header.H)
 	if err != nil {
 		switch err {
 		case ErrAlreadySeen:
+			if hostinfo.multiportRx {
+				// The other host is sending to us with multiport, so only grab the IP
+				via.UdpAddr = netip.AddrPortFrom(via.UdpAddr.Addr(), hostinfo.remote.Port())
+			}
 			// Update remote if preferred
 			if existing.SetRemoteIfPreferred(f.hostMap, via) {
 				// Send a test packet to ensure the other side has also switched to
@@ -357,6 +395,14 @@ func ixHandshakeStage1(f *Interface, via ViaSender, packet []byte, h *header.H)
 			f.messageMetrics.Tx(header.Handshake, header.MessageSubType(msg[1]), 1)
 			if !via.IsRelayed {
 				err := f.outside.WriteTo(msg, via.UdpAddr)
+				if multiportTx {
+					// TODO remove alloc here
+					raw := make([]byte, len(msg)+udp.RawOverhead)
+					copy(raw[udp.RawOverhead:], msg)
+					err = f.udpRaw.WriteTo(raw, udp.RandomSendPort.UDPSendPort(f.multiPort.TxPorts), via.UdpAddr)
+				} else {
+					err = f.outside.WriteTo(msg, via.UdpAddr)
+				}
 				if err != nil {
 					f.l.WithField("vpnAddrs", existing.vpnAddrs).WithField("from", via).
 						WithField("handshake", m{"stage": 2, "style": "ix_psk0"}).WithField("cached", true).
@@ -425,7 +471,14 @@ func ixHandshakeStage1(f *Interface, via ViaSender, packet []byte, h *header.H)
 	// Do the send
 	f.messageMetrics.Tx(header.Handshake, header.MessageSubType(msg[1]), 1)
 	if !via.IsRelayed {
-		err = f.outside.WriteTo(msg, via.UdpAddr)
+		if multiportTx {
+			// TODO remove alloc here
+			raw := make([]byte, len(msg)+udp.RawOverhead)
+			copy(raw[udp.RawOverhead:], msg)
+			err = f.udpRaw.WriteTo(raw, udp.RandomSendPort.UDPSendPort(f.multiPort.TxPorts), via.UdpAddr)
+		} else {
+			err = f.outside.WriteTo(msg, via.UdpAddr)
+		}
 		log := f.l.WithField("vpnAddrs", vpnAddrs).WithField("from", via).
 			WithField("certName", certName).
 			WithField("certVersion", certVersion).
@@ -514,6 +567,20 @@ func ixHandshakeStage2(f *Interface, via ViaSender, hh *HandshakeHostInfo, packe
 		return true
 	}
 
+	if (f.multiPort.Tx || f.multiPort.Rx) && hs.Details.ResponderMultiPort != nil {
+		hostinfo.multiportTx = hs.Details.ResponderMultiPort.RxSupported && f.multiPort.Tx
+		hostinfo.multiportRx = hs.Details.ResponderMultiPort.TxSupported && f.multiPort.Rx
+	}
+
+	if hs.Details.ResponderMultiPort != nil && hs.Details.ResponderMultiPort.BasePort != uint32(via.UdpAddr.Port()) {
+		// The other side sent us a handshake from a different port, make sure
+		// we send responses back to the BasePort
+		via.UdpAddr = netip.AddrPortFrom(
+			via.UdpAddr.Addr(),
+			uint16(hs.Details.ResponderMultiPort.BasePort),
+		)
+	}
+
 	rc, err := cert.Recombine(cert.Version(hs.Details.CertVersion), hs.Details.Cert, ci.H.PeerStatic(), ci.Curve())
 	if err != nil {
 		f.l.WithError(err).WithField("from", via).
@@ -642,7 +709,8 @@ func ixHandshakeStage2(f *Interface, via ViaSender, hh *HandshakeHostInfo, packe
 		WithField("initiatorIndex", hs.Details.InitiatorIndex).WithField("responderIndex", hs.Details.ResponderIndex).
 		WithField("remoteIndex", h.RemoteIndex).WithField("handshake", m{"stage": 2, "style": "ix_psk0"}).
 		WithField("durationNs", duration).
-		WithField("sentCachedPackets", len(hh.packetStore))
+		WithField("sentCachedPackets", len(hh.packetStore)).
+		WithField("multiportTx", hostinfo.multiportTx).WithField("multiportRx", hostinfo.multiportRx)
 	if anyVpnAddrsInCommon {
 		msgRxL.Info("Handshake message received")
 	} else {

handshake_manager.go

@@ -61,6 +61,9 @@ type HandshakeManager struct {
 	f                      *Interface
 	l                      *logrus.Logger
 
+	multiPort MultiPortConfig
+	udpRaw    *udp.RawConn
+
 	// can be used to trigger outbound handshake for the given vpnIp
 	trigger chan netip.Addr
 }
@@ -237,6 +240,7 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
 
 	// Send the handshake to all known ips, stage 2 takes care of assigning the hostinfo.remote based on the first to reply
 	var sentTo []netip.AddrPort
+	var sentMultiport bool
 	hostinfo.remotes.ForEach(hm.mainHostMap.GetPreferredRanges(), func(addr netip.AddrPort, _ bool) {
 		hm.messageMetrics.Tx(header.Handshake, header.MessageSubType(hostinfo.HandshakePacket[0][1]), 1)
 		err := hm.outside.WriteTo(hostinfo.HandshakePacket[0], addr)
@@ -249,6 +253,27 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
 		} else {
 			sentTo = append(sentTo, addr)
 		}
+
+		// Attempt a multiport handshake if we are past the TxHandshakeDelay attempts
+		if hm.multiPort.TxHandshake && hm.udpRaw != nil && hh.counter >= hm.multiPort.TxHandshakeDelay {
+			sentMultiport = true
+			// We need to re-allocate with 8 bytes at the start of SOCK_RAW
+			raw := hostinfo.HandshakePacket[0x80]
+			if raw == nil {
+				raw = make([]byte, len(hostinfo.HandshakePacket[0])+udp.RawOverhead)
+				copy(raw[udp.RawOverhead:], hostinfo.HandshakePacket[0])
+				hostinfo.HandshakePacket[0x80] = raw
+			}
+
+			hm.messageMetrics.Tx(header.Handshake, header.MessageSubType(hostinfo.HandshakePacket[0][1]), 1)
+			err = hm.udpRaw.WriteTo(raw, udp.RandomSendPort.UDPSendPort(hm.multiPort.TxPorts), addr)
+			if err != nil {
+				hostinfo.logger(hm.l).WithField("udpAddr", addr).
+					WithField("initiatorIndex", hostinfo.localIndexId).
+					WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
+					WithError(err).Error("Failed to send handshake message")
+			}
+		}
 	})
 
 	// Don't be too noisy or confusing if we fail to send a handshake - if we don't get through we'll eventually log a timeout,
@@ -257,6 +282,7 @@ func (hm *HandshakeManager) handleOutbound(vpnIp netip.Addr, lighthouseTriggered
 		hostinfo.logger(hm.l).WithField("udpAddrs", sentTo).
 			WithField("initiatorIndex", hostinfo.localIndexId).
 			WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
+			WithField("multiportHandshake", sentMultiport).
 			Info("Handshake message sent")
 	} else if hm.l.Level >= logrus.DebugLevel {
 		hostinfo.logger(hm.l).WithField("udpAddrs", sentTo).

hostmap.go

@@ -243,6 +243,12 @@ type HostInfo struct {
 	networks   *bart.Table[NetworkType]
 	relayState RelayState
 
+	// If true, we should send to this remote using multiport
+	multiportTx bool
+
+	// If true, we will receive from this remote using multiport
+	multiportRx bool
+
 	// HandshakePacket records the packets used to create this hostinfo
 	// We need these to avoid replayed handshake packets creating new hostinfos which causes churn
 	HandshakePacket map[uint8][]byte