filter out magic packets in the middle device to prevent flapping

device/middle_device.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/fosrl/newt/bind"
 	"github.com/fosrl/newt/logger"
 	"golang.zx2c4.com/wireguard/tun"
 )
@@ -423,6 +424,33 @@ func extractDestIP(packet []byte) (netip.Addr, bool) {
 	return netip.Addr{}, false
 }
 
+// extractUDPPayload returns the UDP payload of packet, if packet is a well-formed
+// IPv4 or IPv6 UDP datagram (ignoring IPv6 extension headers).
+func extractUDPPayload(packet []byte) ([]byte, bool) {
+	if len(packet) < 20 {
+		return nil, false
+	}
+
+	const udpProtocol = 17
+
+	switch packet[0] >> 4 {
+	case 4:
+		ihl := int(packet[0]&0x0f) * 4
+		if ihl < 20 || len(packet) < ihl+8 || packet[9] != udpProtocol {
+			return nil, false
+		}
+		return packet[ihl+8:], true
+	case 6:
+		const ipv6HeaderLen = 40
+		if len(packet) < ipv6HeaderLen+8 || packet[6] != udpProtocol {
+			return nil, false
+		}
+		return packet[ipv6HeaderLen+8:], true
+	}
+
+	return nil, false
+}
+
 // Read intercepts packets going UP from the TUN device (towards WireGuard)
 func (d *MiddleDevice) Read(bufs [][]byte, sizes []int, offset int) (n int, err error) {
 	for {
@@ -497,17 +525,19 @@ func (d *MiddleDevice) Read(bufs [][]byte, sizes []int, offset int) (n int, err
 		rules := d.rules
 		d.rulesMutex.RUnlock()
 
-		if len(rules) == 0 {
-			return n, nil
-		}
-
-		// Process packets and filter out handled ones
+		// Process packets and filter out handled ones. This always runs (even with
+		// no per-IP rules registered) so magic connectivity-test packets can be
+		// dropped before they reach WireGuard - see isLeakedMagicPacket.
 		writeIdx := 0
 		for readIdx := 0; readIdx < n; readIdx++ {
 			packet := bufs[readIdx][offset : offset+sizes[readIdx]]
 
+			if isLeakedMagicPacket(packet) {
+				continue
+			}
+
 			destIP, ok := extractDestIP(packet)
-			if !ok {
+			if !ok || len(rules) == 0 {
 				if writeIdx != readIdx {
 					bufs[writeIdx] = bufs[readIdx]
 					sizes[writeIdx] = sizes[readIdx]
@@ -539,6 +569,57 @@ func (d *MiddleDevice) Read(bufs [][]byte, sizes []int, offset int) (n int, err
 	}
 }
 
+// isLeakedMagicPacket reports whether packet carries one of our UDP connectivity-test
+// magic payloads (see bind.IsMagicPacket). These packets are sent directly between
+// physical UDP sockets by the local-endpoint holepunch tester and must never be
+// encapsulated by WireGuard: if OS routing sends one into this TUN interface instead
+// of out the real network interface (e.g. because the destination falls inside a
+// routed tunnel subnet), tunneling and echoing it back would make a LAN-local
+// endpoint falsely appear directly reachable. Dropping it here makes the test
+// correctly time out instead.
+func isLeakedMagicPacket(packet []byte) bool {
+	payload, ok := extractUDPPayload(packet)
+	return ok && bind.IsMagicPacket(payload)
+}
+
+// filterDownstreamBufs drops packets going DOWN to the TUN device (from WireGuard)
+// that are handled by a per-IP rule or are a leaked magic connectivity-test packet
+// (see isLeakedMagicPacket) - always checked, even with no rules registered. It
+// returns bufs unchanged (no allocation) unless a packet actually needs to be
+// dropped, at which point it switches to an owned copy of the buffers kept so far.
+func filterDownstreamBufs(bufs [][]byte, rules []FilterRule, offset int) [][]byte {
+	filtered := bufs
+	for i, buf := range bufs {
+		drop := len(buf) <= offset
+		if !drop {
+			packet := buf[offset:]
+			if isLeakedMagicPacket(packet) {
+				drop = true
+			} else if destIP, ok := extractDestIP(packet); ok && len(rules) > 0 {
+				for _, rule := range rules {
+					if rule.DestIP == destIP && rule.Handler(packet) {
+						drop = true
+						break
+					}
+				}
+			}
+		}
+
+		if drop {
+			if len(filtered) == len(bufs) {
+				// First drop: switch to an owned, growable copy of everything kept so far.
+				filtered = append([][]byte(nil), bufs[:i]...)
+			}
+			continue
+		}
+
+		if len(filtered) != len(bufs) {
+			filtered = append(filtered, buf)
+		}
+	}
+	return filtered
+}
+
 // Write intercepts packets going DOWN to the TUN device (from WireGuard)
 func (d *MiddleDevice) Write(bufs [][]byte, offset int) (int, error) {
 	for {
@@ -558,38 +639,7 @@ func (d *MiddleDevice) Write(bufs [][]byte, offset int) (int, error) {
 		rules := d.rules
 		d.rulesMutex.RUnlock()
 
-		var filteredBufs [][]byte
-		if len(rules) == 0 {
-			filteredBufs = bufs
-		} else {
-			filteredBufs = make([][]byte, 0, len(bufs))
-			for _, buf := range bufs {
-				if len(buf) <= offset {
-					continue
-				}
-
-				packet := buf[offset:]
-				destIP, ok := extractDestIP(packet)
-				if !ok {
-					filteredBufs = append(filteredBufs, buf)
-					continue
-				}
-
-				handled := false
-				for _, rule := range rules {
-					if rule.DestIP == destIP {
-						if rule.Handler(packet) {
-							handled = true
-							break
-						}
-					}
-				}
-
-				if !handled {
-					filteredBufs = append(filteredBufs, buf)
-				}
-			}
-		}
+		filteredBufs := filterDownstreamBufs(bufs, rules, offset)
 
 		if len(filteredBufs) == 0 {
 			return len(bufs), nil

device/middle_device_test.go

@@ -4,9 +4,22 @@ import (
 	"net/netip"
 	"testing"
 
+	"github.com/fosrl/newt/bind"
 	"github.com/fosrl/newt/util"
 )
 
+// buildIPv4UDPPacket builds a minimal IPv4/UDP packet (no options) carrying payload.
+func buildIPv4UDPPacket(payload []byte) []byte {
+	const ipHeaderLen = 20
+	const udpHeaderLen = 8
+
+	packet := make([]byte, ipHeaderLen+udpHeaderLen+len(payload))
+	packet[0] = 0x45 // version 4, IHL 5
+	packet[9] = 17    // protocol: UDP
+	copy(packet[ipHeaderLen+udpHeaderLen:], payload)
+	return packet
+}
+
 func TestExtractDestIP(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -88,6 +101,49 @@ func TestGetProtocol(t *testing.T) {
 	}
 }
 
+func TestIsLeakedMagicPacket(t *testing.T) {
+	request := make([]byte, bind.MagicTestRequestLen)
+	copy(request, bind.MagicTestRequest)
+
+	response := make([]byte, bind.MagicTestResponseLen)
+	copy(response, bind.MagicTestResponse)
+
+	tests := []struct {
+		name   string
+		packet []byte
+		want   bool
+	}{
+		{
+			name:   "magic test request leaked into tunnel",
+			packet: buildIPv4UDPPacket(request),
+			want:   true,
+		},
+		{
+			name:   "magic test response leaked into tunnel",
+			packet: buildIPv4UDPPacket(response),
+			want:   true,
+		},
+		{
+			name:   "ordinary UDP payload",
+			packet: buildIPv4UDPPacket([]byte("just some ordinary application data")),
+			want:   false,
+		},
+		{
+			name:   "too short to be a packet",
+			packet: []byte{0x45, 0x00},
+			want:   false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isLeakedMagicPacket(tt.packet); got != tt.want {
+				t.Errorf("isLeakedMagicPacket() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func BenchmarkExtractDestIP(b *testing.B) {
 	packet := []byte{
 		0x45, 0x00, 0x00, 0x54, 0x00, 0x00, 0x40, 0x00,
@@ -100,3 +156,61 @@ func BenchmarkExtractDestIP(b *testing.B) {
 		extractDestIP(packet)
 	}
 }
+
+func TestFilterDownstreamBufsNoDropIsAllocFree(t *testing.T) {
+	bufs := make([][]byte, 128)
+	for i := range bufs {
+		bufs[i] = buildIPv4UDPPacket(make([]byte, 1372))
+	}
+
+	allocs := testing.AllocsPerRun(1000, func() {
+		out := filterDownstreamBufs(bufs, nil, 0)
+		if len(out) != len(bufs) {
+			t.Fatalf("expected no packets dropped, got %d/%d", len(out), len(bufs))
+		}
+	})
+
+	if allocs != 0 {
+		t.Errorf("filterDownstreamBufs() with nothing to drop allocated %v times per call, want 0", allocs)
+	}
+}
+
+func TestFilterDownstreamBufsDropsMagicPacket(t *testing.T) {
+	request := make([]byte, bind.MagicTestRequestLen)
+	copy(request, bind.MagicTestRequest)
+
+	bufs := [][]byte{
+		buildIPv4UDPPacket([]byte("ordinary payload one")),
+		buildIPv4UDPPacket(request),
+		buildIPv4UDPPacket([]byte("ordinary payload two")),
+	}
+
+	out := filterDownstreamBufs(bufs, nil, 0)
+	if len(out) != 2 {
+		t.Fatalf("expected 1 packet dropped, got %d remaining", len(out))
+	}
+}
+
+func BenchmarkFilterDownstreamBufsNoDrop(b *testing.B) {
+	bufs := make([][]byte, 128)
+	for i := range bufs {
+		bufs[i] = buildIPv4UDPPacket(make([]byte, 1372))
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		filterDownstreamBufs(bufs, nil, 0)
+	}
+}
+
+func BenchmarkIsLeakedMagicPacket(b *testing.B) {
+	// A typical ~1400 byte ordinary application payload (the common case on the
+	// hot path - almost every real packet should look like this).
+	ordinary := buildIPv4UDPPacket(make([]byte, 1372))
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		isLeakedMagicPacket(ordinary)
+	}
+}