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

@@ -218,6 +218,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

@@ -244,6 +244,47 @@ tun:
   # in nebula configuration files. Default false, not reloadable.
   #use_system_route_table: false
 
+  # 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
+
 # TODO
 # Configure logging level
 logging:

firewall/packet.go

@@ -3,6 +3,7 @@ package firewall
 import (
 	"encoding/json"
 	"fmt"
+	mathrand "math/rand"
 	"net/netip"
 )
 
@@ -59,3 +60,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

@@ -7,6 +7,7 @@ import (
 	"github.com/flynn/noise"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/header"
+	"github.com/slackhq/nebula/udp"
 )
 
 // NOISE IX Handshakes
@@ -31,6 +32,15 @@ func ixHandshakeStage0(f *Interface, hh *HandshakeHostInfo) bool {
 		Cert:           certState.RawCertificateNoKey,
 	}
 
+	if f.multiPort.Tx || f.multiPort.Rx {
+		hsProto.InitiatorMultiPort = &MultiPortDetails{
+			RxSupported: f.multiPort.Rx,
+			TxSupported: f.multiPort.Tx,
+			BasePort:    uint32(f.multiPort.TxBasePort),
+			TotalPorts:  uint32(f.multiPort.TxPorts),
+		}
+	}
+
 	hsBytes := []byte{}
 
 	hs := &NebulaHandshake{
@@ -143,13 +153,35 @@ func ixHandshakeStage1(f *Interface, addr netip.AddrPort, via *ViaSender, packet
 		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(addr.Port()) {
+		// The other side sent us a handshake from a different port, make sure
+		// we send responses back to the BasePort
+		addr = netip.AddrPortFrom(addr.Addr(), uint16(hs.Details.InitiatorMultiPort.BasePort))
+	}
+
 	hostinfo := &HostInfo{
 		ConnectionState:   ci,
 		localIndexId:      myIndex,
 		remoteIndexId:     hs.Details.InitiatorIndex,
 		vpnIp:             vpnIp,
 		HandshakePacket:   make(map[uint8][]byte, 0),
 		lastHandshakeTime: hs.Details.Time,
+		multiportTx:       multiportTx,
+		multiportRx:       multiportRx,
 		relayState: RelayState{
 			relays:        map[netip.Addr]struct{}{},
 			relayForByIp:  map[netip.Addr]*Relay{},
@@ -163,6 +195,7 @@ func ixHandshakeStage1(f *Interface, addr netip.AddrPort, via *ViaSender, packet
 		WithField("issuer", issuer).
 		WithField("initiatorIndex", hs.Details.InitiatorIndex).WithField("responderIndex", hs.Details.ResponderIndex).
 		WithField("remoteIndex", h.RemoteIndex).WithField("handshake", m{"stage": 1, "style": "ix_psk0"}).
+		WithField("multiportTx", multiportTx).WithField("multiportRx", multiportRx).
 		Info("Handshake message received")
 
 	hs.Details.ResponderIndex = myIndex
@@ -222,6 +255,10 @@ func ixHandshakeStage1(f *Interface, addr netip.AddrPort, via *ViaSender, packet
 	if err != nil {
 		switch err {
 		case ErrAlreadySeen:
+			if hostinfo.multiportRx {
+				// The other host is sending to us with multiport, so only grab the IP
+				addr = netip.AddrPortFrom(addr.Addr(), hostinfo.remote.Port())
+			}
 			// Update remote if preferred
 			if existing.SetRemoteIfPreferred(f.hostMap, addr) {
 				// Send a test packet to ensure the other side has also switched to
@@ -232,7 +269,14 @@ func ixHandshakeStage1(f *Interface, addr netip.AddrPort, via *ViaSender, packet
 			msg = existing.HandshakePacket[2]
 			f.messageMetrics.Tx(header.Handshake, header.MessageSubType(msg[1]), 1)
 			if addr.IsValid() {
-				err := f.outside.WriteTo(msg, addr)
+				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), addr)
+				} else {
+					err = f.outside.WriteTo(msg, addr)
+				}
 				if err != nil {
 					f.l.WithField("vpnIp", existing.vpnIp).WithField("udpAddr", addr).
 						WithField("handshake", m{"stage": 2, "style": "ix_psk0"}).WithField("cached", true).
@@ -298,7 +342,14 @@ func ixHandshakeStage1(f *Interface, addr netip.AddrPort, via *ViaSender, packet
 	// Do the send
 	f.messageMetrics.Tx(header.Handshake, header.MessageSubType(msg[1]), 1)
 	if addr.IsValid() {
-		err = f.outside.WriteTo(msg, addr)
+		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), addr)
+		} else {
+			err = f.outside.WriteTo(msg, addr)
+		}
 		if err != nil {
 			f.l.WithField("vpnIp", vpnIp).WithField("udpAddr", addr).
 				WithField("certName", certName).
@@ -387,6 +438,20 @@ func ixHandshakeStage2(f *Interface, addr netip.AddrPort, via *ViaSender, hh *Ha
 		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(addr.Port()) {
+		// The other side sent us a handshake from a different port, make sure
+		// we send responses back to the BasePort
+		addr = netip.AddrPortFrom(
+			addr.Addr(),
+			uint16(hs.Details.ResponderMultiPort.BasePort),
+		)
+	}
+
 	remoteCert, err := RecombineCertAndValidate(ci.H, hs.Details.Cert, f.pki.GetCAPool())
 	if err != nil {
 		e := f.l.WithError(err).WithField("vpnIp", hostinfo.vpnIp).WithField("udpAddr", addr).
@@ -468,6 +533,7 @@ func ixHandshakeStage2(f *Interface, addr netip.AddrPort, via *ViaSender, hh *Ha
 		WithField("remoteIndex", h.RemoteIndex).WithField("handshake", m{"stage": 2, "style": "ix_psk0"}).
 		WithField("durationNs", duration).
 		WithField("sentCachedPackets", len(hh.packetStore)).
+		WithField("multiportTx", hostinfo.multiportTx).WithField("multiportRx", hostinfo.multiportRx).
 		Info("Handshake message received")
 
 	hostinfo.remoteIndexId = hs.Details.ResponderIndex

handshake_manager.go

@@ -60,6 +60,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
 }
@@ -235,6 +238,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)
@@ -247,6 +251,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,
@@ -255,6 +280,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.IsLevelEnabled(logrus.DebugLevel) {
 		hostinfo.logger(hm.l).WithField("udpAddrs", sentTo).

hostmap.go

@@ -195,6 +195,12 @@ type HostInfo struct {
 	remoteCidr      *bart.Table[struct{}]
 	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