plugin/pkg/proxyproto: preserve the UDP peer for LOCAL (#8581)

This PR consumes LOCAL headers without replacing or caching the actual UDP peer address.

Signed-off-by: Yong Tang <yong.tang.github@outlook.com>

plugin/pkg/proxyproto/proxyproto.go

@@ -130,21 +130,23 @@ func (c *PacketConn) readFrom(p []byte, addr net.Addr) (_ int, _ net.Addr, err e
 		fallthrough
 	case proxyproto.USE:
 		if header != nil {
-			addr = &Addr{u: addr, r: header.SourceAddr}
+			if !header.Command.IsLocal() {
+				addr = &Addr{u: addr, r: header.SourceAddr}
 
-			if c.UDPSessionTrackingTTL > 0 {
+				if c.UDPSessionTrackingTTL > 0 {
-				// Cache the real source address for subsequent headerless datagrams.
+					// Cache the real source address for subsequent headerless datagrams.
-				// Spectrum sends the header in a standalone datagram with no DNS
+					// Spectrum sends the header in a standalone datagram with no DNS
-				// payload; refresh or insert the entry either way so that the TTL
+					// payload; refresh or insert the entry either way so that the TTL
-				// resets on every header packet.
+					// resets on every header packet.
-				c.storeSession(addr.(*Addr).u, header)
+					c.storeSession(addr.(*Addr).u, header)
-
-				if len(payload) == 0 {
-					// Header-only datagram: no DNS payload to return; loop back
-					// to read the next datagram.
-					return 0, nil, errHeaderOnly
 				}
 			}
+
+			if c.UDPSessionTrackingTTL > 0 && len(payload) == 0 {
+				// Header-only datagram: no DNS payload to return; loop back
+				// to read the next datagram.
+				return 0, nil, errHeaderOnly
+			}
 		} else if c.UDPSessionTrackingTTL > 0 {
 			// No header present – look for a cached header for this remote.
 			if cachedHeader, ok := c.lookupSession(addr); ok {

plugin/pkg/proxyproto/udp_session_test.go

@@ -1,6 +1,7 @@
 package proxyproto
 
 import (
+	"bytes"
 	"io"
 	"net"
 	"testing"
@@ -134,6 +135,43 @@ func TestStoreSessionEvictsOldest(t *testing.T) {
 	}
 }
 
+func TestPacketConnLocalCommandUsesPeerAddress(t *testing.T) {
+	payload := []byte{1, 2, 3, 4}
+	packet := append([]byte{
+		0x0d, 0x0a, 0x0d, 0x0a, 0x00, 0x0d, 0x0a, 0x51,
+		0x55, 0x49, 0x54, 0x0a, // PPv2 signature
+		0x20,       // version 2, LOCAL command
+		0x12,       // UDPv4
+		0x00, 0x0c, // address length
+		192, 0, 2, 1,
+		192, 0, 2, 53,
+		0x30, 0x39,
+		0x00, 0x35,
+	}, payload...)
+	peer := udpAddr("198.51.100.1", 53000)
+	pc := &PacketConn{
+		PacketConn: &singlePacketConn{},
+		ConnPolicy: func(proxyproto.ConnPolicyOptions) (proxyproto.Policy, error) {
+			return proxyproto.USE, nil
+		},
+		UDPSessionTrackingTTL: time.Minute,
+	}
+
+	n, addr, err := pc.readFrom(packet, peer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if addr.String() != peer.String() {
+		t.Fatalf("LOCAL command changed peer address from %s to %s", peer, addr)
+	}
+	if !bytes.Equal(packet[:n], payload) {
+		t.Fatalf("payload = %v, want %v", packet[:n], payload)
+	}
+	if _, ok := pc.lookupSession(peer); ok {
+		t.Fatal("LOCAL command stored a spoofed UDP session")
+	}
+}
+
 func TestPacketConnReadFromMalformedPPv2NonUDPDoesNotPanic(t *testing.T) {
 	pc := &singlePacketConn{
 		packet: []byte{