From 3549fe5ad8a6e9b863b3e4f7b9bf738facff87ee Mon Sep 17 00:00:00 2001 From: tsosunchia <59512455+tsosunchia@users.noreply.github.com> Date: Sat, 24 Oct 2026 18:35:22 +0800 Subject: [PATCH] fix(udp): preserve wildcard reply sources and isolate destination sessions (#565) * fix(udp): preserve local destinations and unicast reply sources on Linux * fix(udp): select unicast sources for IPv6 multicast replies * fix(udp): retain interface zones for link-local multicast sessions * fix(udp): isolate interface-local multicast sessions * fix(udp): include ingress interfaces in session identity * fix(udp): restrict destination zones to IPv6 * ci(udp): follow rc5 patched Go toolchain requirement --------- Co-authored-by: sagit <36596628+Sagit-chu@users.noreply.github.com> --- .github/workflows/udp-destination.yml | 87 +++++ docs/udp-wildcard-source.md | 133 ++++++++ go-gost/x/internal/net/udp/listener.go | 24 +- go-gost/x/internal/net/udp/packet.go | 12 + go-gost/x/internal/net/udp/packet_linux.go | 116 +++++++ .../x/internal/net/udp/packet_linux_test.go | 304 ++++++++++++++++++ go-gost/x/internal/net/udp/packet_other.go | 9 + .../x/listener/udp/destination_linux_test.go | 289 +++++++++++++++++ go-gost/x/listener/udp/listener.go | 8 +- 9 files changed, 977 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/udp-destination.yml create mode 100644 docs/udp-wildcard-source.md create mode 100644 go-gost/x/internal/net/udp/packet.go create mode 100644 go-gost/x/internal/net/udp/packet_linux.go create mode 100644 go-gost/x/internal/net/udp/packet_linux_test.go create mode 100644 go-gost/x/internal/net/udp/packet_other.go create mode 100644 go-gost/x/listener/udp/destination_linux_test.go diff --git a/.github/workflows/udp-destination.yml b/.github/workflows/udp-destination.yml new file mode 100644 index 0000000..5fefb33 --- /dev/null +++ b/.github/workflows/udp-destination.yml @@ -0,0 +1,87 @@ +name: UDP Destination Regression + +on: + push: + branches: ['main', 'fix/udp-wildcard-source', 'pr/udp-wildcard-source'] + pull_request: + paths: + - 'go-gost/**' + - '.github/workflows/udp-destination.yml' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: udp-destination-${{ github.ref }} + cancel-in-progress: true + +jobs: + linux: + runs-on: ubuntu-24.04 + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version-file: 'go-gost/go.mod' + cache-dependency-path: | + go-gost/go.sum + go-gost/x/go.sum + + - name: Test agent module + working-directory: go-gost + run: go test ./... + + - name: Test protocol module + working-directory: go-gost/x + run: go test ./... + + - name: Check UDP races + working-directory: go-gost/x + run: go test -race ./internal/net/udp ./listener/udp + + - name: Vet UDP implementation + working-directory: go-gost/x + run: go vet ./internal/net/udp ./listener/udp + + - name: Build destination regression executable + working-directory: go-gost/x + run: | + go test -c -race -o "$RUNNER_TEMP/udp-destination.test" ./listener/udp + go test -c -race -o "$RUNNER_TEMP/udp-packet.test" ./internal/net/udp + + - name: Test multiple destinations in isolated network namespace + run: | + sudo unshare --net -- sh -eu -c ' + sysctl -qw net.ipv6.conf.all.disable_ipv6=0 + ip link set lo up + ip addr add 198.18.0.1/32 dev lo + ip addr add 169.254.1.2/32 dev lo + ip -6 addr add fd00:1::1/128 dev lo nodad + ip -6 addr add fd00:2::1/128 dev lo nodad + ip -6 addr add fd00:3::1/128 dev lo nodad + for suffix in "" 2; do + ip link add "mc-tx$suffix" type veth peer name "mc-rx$suffix" + ip link set "mc-tx$suffix" addrgenmode none up + ip link set "mc-rx$suffix" addrgenmode none up + ip -6 addr add fe80::1/64 dev "mc-tx$suffix" nodad + ip -6 addr add fe80::2/64 dev "mc-rx$suffix" nodad + done + ip addr add 198.18.1.1/24 dev mc-tx + ip addr add 198.18.2.1/24 dev mc-tx2 + export FLUX_TEST_IPV6_ADDRS=fd00:1::1,fd00:2::1,fd00:3::1 + export FLUX_TEST_IPV4_LINK_LOCAL=169.254.1.2 + export FLUX_TEST_MULTICAST_RX=mc-rx FLUX_TEST_MULTICAST_TX=mc-tx + export FLUX_TEST_MULTICAST_RX2=mc-rx2 FLUX_TEST_MULTICAST_TX2=mc-tx2 + "$1" -test.v -test.timeout=60s -test.count=5 + exec "$2" -test.v -test.timeout=60s -test.count=5 + ' sh "$RUNNER_TEMP/udp-destination.test" "$RUNNER_TEMP/udp-packet.test" + + - name: Build Linux agent + working-directory: go-gost + env: + CGO_ENABLED: '0' + GOOS: linux + GOARCH: amd64 + run: go build -trimpath -o "$RUNNER_TEMP/flux-agent" . diff --git a/docs/udp-wildcard-source.md b/docs/udp-wildcard-source.md new file mode 100644 index 0000000..002e919 --- /dev/null +++ b/docs/udp-wildcard-source.md @@ -0,0 +1,133 @@ +# Preserve UDP reply source addresses on Linux + +The Linux UDP listener preserves the destination address of each incoming +datagram so that replies use the correct local source. This change was developed +against FLVX `3.0.1-rc4`, commit +`d9303032cad09b33cd0b7c91704fd671fbfdd078`. +It also includes the `3.0.1-rc5` runtime update at +`c794586cc7773e5f967783895d3071538442bbf7`. + +## Problem + +A wildcard UDP socket bound to `0.0.0.0:port` or `[::]:port` can receive datagrams +addressed to several local IP addresses. Previously, the listener retained only +the client address. A reply sent through `WriteTo` could therefore use a source +IP chosen by the kernel instead of the IP the client contacted. DNS clients, for +example, can reject such replies as coming from an unexpected source. + +The UDP session pool was also keyed only by the client address and port. When +one client socket contacted multiple local addresses on the same listener port, +these requests could share a session despite having different destinations. + +## Changes + +- Linux wildcard UDP sockets enable IPv4/IPv6 packet information. `recvmsg` + preserves the datagram's local destination and interface; `sendmsg` uses the + corresponding source address and interface when replying. +- IPv4 keeps the original destination (`ipi_addr`) separate from the local + reply address (`ipi_spec_dst`), so broadcast requests receive unicast replies + rather than attempting to send with a broadcast source address. +- IPv6 multicast destinations remain part of session identity, but replies + leave the source unspecified and retain the received interface index so the + kernel can select a valid local unicast source. +- The destination metadata travels through existing metrics, statistics, + admission and traffic-limiter wrappers. The exposed remote address remains + the real client address. +- Sessions are keyed by the local destination endpoint, remote client endpoint + and received interface index. Identical IPv4 broadcast or multicast endpoints + on different interfaces therefore retain separate reply control information. + An accepted session reports its actual local destination. + Link-local IPv6 unicast and link-local or interface-local multicast + destinations retain their interface zone, keeping identical multicast groups + on different interfaces separate even when the client uses the same + non-link-local IP and source port. + IPv4 destinations, including link-local unicast and multicast, never receive + an IPv6 zone; their interface isolation uses the separate index in the key. +- Explicitly bound UDP sockets retain their existing behavior. Non-Linux + platforms retain their existing implementation. No configuration keys or + panel API changes are required. + +The fix affects the shared UDP listener, including services dynamically created +after startup. It does not configure listening addresses, NAT, routing or +firewalls. Multi-homed hosts still need correct source policy routing for their +uplinks; this patch does not fix SSH/TCP routing or packet loss before ingress. + +## Build + +Use Go 1.26.10 or a compatible newer toolchain. From the repository root: + +```sh +cd go-gost +CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath \ + -ldflags '-s -w' -o flux_agent . +``` + +Use the agent with the normal panel integration and forwarding configuration. +Selecting a wildcard listener enables destination preservation automatically +on Linux. The patch does not select a listener address or change the upstream +connection's outbound interface. + +## Regression tests + +The Linux destination tests cover IPv4 wildcard and dual-stack wildcard sockets, +explicit IPv4/IPv6 bindings, one client source port contacting multiple local +destinations, repeated requests, session expiry, and active metrics, statistics, +admission and traffic-limit wrappers. A separate IPv6 case binds the client to +`::1` and checks the exact remote IP and port for every destination. Broadcast +tests exercise both IPv4 and dual-stack listeners, retaining the broadcast +destination while checking that replies use a local unicast address. +An IPv6 multicast regression sends across an isolated veth pair and checks the +multicast session destination and the unicast reply, including interface zones. +A two-interface case uses one fixed ULA client endpoint to contact the same +link-local or interface-local multicast group on both interfaces and checks +distinct sessions and their individual reply-interface metadata. +IPv4 and dual-stack cases likewise use a single fixed client endpoint to send +to one multicast group or the limited broadcast address on two interfaces, +checking separate sessions and their reply-interface indices. +Link-local IPv4 unicast and multicast regressions also check that accepted +local endpoints have no IPv6 zone, on both IPv4 and dual-stack listeners. + +Run the standard suites on Linux: + +```sh +(cd go-gost && go test ./...) +(cd go-gost/x && go test ./...) +``` + +To exercise multiple IPv6 addresses without modifying the host's live network, +compile the tests first and run them inside a disposable network namespace. +The following commands require Linux, root access for `unshare`, and `iproute2`: + +```sh +cd go-gost/x +go test -c -race -o /tmp/flvx-udp-destination.test ./listener/udp +go test -c -race -o /tmp/flvx-udp-packet.test ./internal/net/udp +sudo unshare --net -- sh -eu -c ' + sysctl -qw net.ipv6.conf.all.disable_ipv6=0 + ip link set lo up + ip addr add 198.18.0.1/32 dev lo + ip addr add 169.254.1.2/32 dev lo + ip -6 addr add fd00:1::1/128 dev lo nodad + ip -6 addr add fd00:2::1/128 dev lo nodad + ip -6 addr add fd00:3::1/128 dev lo nodad + for suffix in "" 2; do + ip link add "mc-tx$suffix" type veth peer name "mc-rx$suffix" + ip link set "mc-tx$suffix" addrgenmode none up + ip link set "mc-rx$suffix" addrgenmode none up + ip -6 addr add fe80::1/64 dev "mc-tx$suffix" nodad + ip -6 addr add fe80::2/64 dev "mc-rx$suffix" nodad + done + ip addr add 198.18.1.1/24 dev mc-tx + ip addr add 198.18.2.1/24 dev mc-tx2 + export FLUX_TEST_IPV6_ADDRS=fd00:1::1,fd00:2::1,fd00:3::1 + export FLUX_TEST_IPV4_LINK_LOCAL=169.254.1.2 + export FLUX_TEST_MULTICAST_RX=mc-rx FLUX_TEST_MULTICAST_TX=mc-tx + export FLUX_TEST_MULTICAST_RX2=mc-rx2 FLUX_TEST_MULTICAST_TX2=mc-tx2 + /tmp/flvx-udp-destination.test -test.v -test.timeout=60s -test.count=5 + exec /tmp/flvx-udp-packet.test -test.v -test.timeout=60s -test.count=5 +' +``` + +The addresses above are isolated test addresses, not deployment settings. +The `UDP Destination Regression` GitHub Actions workflow runs these namespace +tests, the race detector, both agent module suites, and a Linux agent build. diff --git a/go-gost/x/internal/net/udp/listener.go b/go-gost/x/internal/net/udp/listener.go index fac064e..6362599 100644 --- a/go-gost/x/internal/net/udp/listener.go +++ b/go-gost/x/internal/net/udp/listener.go @@ -3,6 +3,7 @@ package udp import ( "errors" "net" + "strconv" "sync" "sync/atomic" "time" @@ -73,6 +74,7 @@ func (ln *listener) listenLoop() { n, raddr, err := ln.conn.ReadFrom(b) if err != nil { + bufpool.Put(b) ln.errChan <- err close(ln.errChan) return @@ -110,15 +112,27 @@ func (ln *listener) Close() error { } func (ln *listener) getConn(raddr net.Addr) *conn { - c, ok := ln.connPool.Get(raddr.String()) + laddr := ln.Addr() + peer := raddr + var packet *packetAddr + if received, ok := raddr.(*packetAddr); ok { + packet = received + laddr, peer = received.local, received.Addr + } + key := laddr.String() + "|" + peer.String() + if packet != nil { + key += "|" + strconv.Itoa(packet.ifIndex) + } + c, ok := ln.connPool.Get(key) if ok && !c.isClosed() { return c } - c = newConn(ln.conn, ln.Addr(), raddr, ln.config.ReadQueueSize, ln.config.Keepalive) + c = newConn(ln.conn, laddr, peer, ln.config.ReadQueueSize, ln.config.Keepalive) + c.packet = packet select { case ln.cqueue <- c: - ln.connPool.Set(raddr.String(), c) + ln.connPool.Set(key, c) return c default: c.Close() @@ -132,6 +146,7 @@ type conn struct { net.PacketConn localAddr net.Addr remoteAddr net.Addr + packet *packetAddr rc chan []byte // data receive queue idle int32 // indicate the connection is idle closed chan struct{} @@ -176,6 +191,9 @@ func (c *conn) WriteTo(b []byte, addr net.Addr) (n int, err error) { if !c.keepalive { defer c.Close() } + if c.packet != nil { + addr = &packetAddr{Addr: addr, local: c.packet.local, control: c.packet.control, ifIndex: c.packet.ifIndex} + } return c.PacketConn.WriteTo(b, addr) } diff --git a/go-gost/x/internal/net/udp/packet.go b/go-gost/x/internal/net/udp/packet.go new file mode 100644 index 0000000..738fd19 --- /dev/null +++ b/go-gost/x/internal/net/udp/packet.go @@ -0,0 +1,12 @@ +package udp + +import "net" + +// packetAddr carries the received destination through the PacketConn wrappers. +// Its public address remains the real peer for admission, metrics and limits. +type packetAddr struct { + net.Addr + local *net.UDPAddr + control []byte + ifIndex int +} diff --git a/go-gost/x/internal/net/udp/packet_linux.go b/go-gost/x/internal/net/udp/packet_linux.go new file mode 100644 index 0000000..3b3899b --- /dev/null +++ b/go-gost/x/internal/net/udp/packet_linux.go @@ -0,0 +1,116 @@ +package udp + +import ( + "encoding/binary" + "errors" + "net" + + "golang.org/x/net/ipv6" + "golang.org/x/sys/unix" +) + +// PreserveDestination pins wildcard UDP replies to the address they arrived on. +func PreserveDestination(c *net.UDPConn) (net.PacketConn, error) { + local := c.LocalAddr().(*net.UDPAddr) + if !local.IP.IsUnspecified() { + return c, nil + } + raw, err := c.SyscallConn() + if err != nil { + return nil, err + } + var sockErr error + if err = raw.Control(func(fd uintptr) { + if local.IP.To4() != nil { + sockErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_PKTINFO, 1) + return + } + if sockErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_RECVPKTINFO, 1); sockErr != nil { + return + } + var onlyV6 int + onlyV6, sockErr = unix.GetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_V6ONLY) + if sockErr == nil && onlyV6 == 0 { + sockErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_PKTINFO, 1) + } + }); err != nil { + return nil, err + } + if sockErr != nil { + return nil, sockErr + } + return &destinationConn{UDPConn: c}, nil +} + +type destinationConn struct { + *net.UDPConn +} + +func (c *destinationConn) ReadFrom(b []byte) (int, net.Addr, error) { + var oob [128]byte + n, oobn, flags, remote, err := c.ReadMsgUDP(b, oob[:]) + if err != nil { + return 0, remote, err + } + if flags&unix.MSG_CTRUNC != 0 { + return 0, nil, errors.New("UDP destination control message truncated") + } + var dst net.IP + var control []byte + var ifIndex int + if remote.IP.To4() != nil { + messages, err := unix.ParseSocketControlMessage(oob[:oobn]) + if err != nil { + return 0, nil, err + } + for _, message := range messages { + if message.Header.Level != unix.IPPROTO_IP || message.Header.Type != unix.IP_PKTINFO { + continue + } + var info unix.Inet4Pktinfo + if _, err := binary.Decode(message.Data, binary.NativeEndian, &info); err != nil { + return 0, nil, err + } + dst, ifIndex = net.IP(info.Addr[:]), int(info.Ifindex) + // Spec_dst remains a local unicast address for broadcast destinations. + control = unix.PktInfo4(&unix.Inet4Pktinfo{Ifindex: info.Ifindex, Spec_dst: info.Spec_dst}) + break + } + } else { + var cm ipv6.ControlMessage + if err = cm.Parse(oob[:oobn]); err != nil { + return 0, nil, err + } + dst, ifIndex = cm.Dst, cm.IfIndex + source := dst + if dst.IsMulticast() { + source = nil + } + control = (&ipv6.ControlMessage{Src: source, IfIndex: ifIndex}).Marshal() + } + if len(dst) == 0 || dst.IsUnspecified() { + return 0, nil, errors.New("UDP destination address missing") + } + local := &net.UDPAddr{IP: append(net.IP(nil), dst...), Port: c.LocalAddr().(*net.UDPAddr).Port} + if dst.To4() == nil && (dst.IsLinkLocalUnicast() || dst.IsLinkLocalMulticast() || dst.IsInterfaceLocalMulticast()) { + if iface, lookupErr := net.InterfaceByIndex(ifIndex); lookupErr == nil { + local.Zone = iface.Name + } + } + return n, &packetAddr{Addr: remote, local: local, control: control, ifIndex: ifIndex}, nil +} + +func (c *destinationConn) WriteTo(b []byte, addr net.Addr) (int, error) { + if packet, ok := addr.(*packetAddr); ok { + remote, ok := packet.Addr.(*net.UDPAddr) + if !ok { + return 0, errors.New("invalid UDP peer address") + } + n, _, err := c.WriteMsgUDP(b, packet.control, remote) + if n < 0 { + n = 0 + } + return n, err + } + return c.UDPConn.WriteTo(b, addr) +} diff --git a/go-gost/x/internal/net/udp/packet_linux_test.go b/go-gost/x/internal/net/udp/packet_linux_test.go new file mode 100644 index 0000000..639df8e --- /dev/null +++ b/go-gost/x/internal/net/udp/packet_linux_test.go @@ -0,0 +1,304 @@ +package udp + +import ( + "encoding/binary" + "fmt" + "io" + "net" + "os" + "testing" + "time" + + "golang.org/x/net/ipv4" + "golang.org/x/net/ipv6" + "golang.org/x/sys/unix" + + xlogger "github.com/go-gost/x/logger" +) + +func TestMulticastDestinationRepliesFromUnicast(t *testing.T) { + rxName, txName := os.Getenv("FLUX_TEST_MULTICAST_RX"), os.Getenv("FLUX_TEST_MULTICAST_TX") + if rxName == "" || txName == "" { + t.Skip("requires isolated multicast veth interfaces") + } + rx, err := net.InterfaceByName(rxName) + if err != nil { + t.Fatal(err) + } + tx, err := net.InterfaceByName(txName) + if err != nil { + t.Fatal(err) + } + group := net.ParseIP("ff02::114") + socket, err := net.ListenMulticastUDP("udp6", rx, &net.UDPAddr{IP: group}) + if err != nil { + t.Fatal(err) + } + defer socket.Close() + pc, err := PreserveDestination(socket) + if err != nil { + t.Fatal(err) + } + ln := NewListener(pc, &ListenConfig{ + Backlog: 1, ReadQueueSize: 1, ReadBufferSize: 128, TTL: time.Second, Keepalive: true, + Logger: xlogger.NewLogger(xlogger.OutputOption(io.Discard)), + }) + defer ln.Close() + client, err := net.ListenUDP("udp6", &net.UDPAddr{IP: net.ParseIP("fe80::1"), Zone: txName}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + if err := ipv6.NewPacketConn(client).SetMulticastInterface(tx); err != nil { + t.Fatal(err) + } + client.SetDeadline(time.Now().Add(3 * time.Second)) + port := socket.LocalAddr().(*net.UDPAddr).Port + serverResult := make(chan error, 1) + go func() { + c, err := ln.Accept() + if err != nil { + serverResult <- err + return + } + defer c.Close() + local, peer := c.LocalAddr().(*net.UDPAddr), c.RemoteAddr().(*net.UDPAddr) + if !local.IP.Equal(group) || local.Port != port || local.Zone != rxName || !peer.IP.Equal(net.ParseIP("fe80::1")) || + peer.Port != client.LocalAddr().(*net.UDPAddr).Port || peer.Zone != rxName { + serverResult <- fmt.Errorf("unexpected multicast session: local=%s peer=%s", local, peer) + return + } + var buf [128]byte + n, err := c.Read(buf[:]) + if err == nil { + _, err = c.Write(buf[:n]) + } + serverResult <- err + }() + if _, err := client.WriteToUDP([]byte("multicast"), &net.UDPAddr{IP: group, Port: port, Zone: txName}); err != nil { + t.Fatal(err) + } + select { + case err := <-serverResult: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("multicast session timed out") + } + var buf [128]byte + n, from, err := client.ReadFromUDP(buf[:]) + if err != nil { + t.Fatal(err) + } + if !from.IP.Equal(net.ParseIP("fe80::2")) || from.Port != port || from.Zone != txName || string(buf[:n]) != "multicast" { + t.Fatalf("unexpected multicast reply %q from %s", buf[:n], from) + } +} + +func TestLinkLocalMulticastSessionsRetainZones(t *testing.T) { + testScopedMulticastSessions(t, "ff02::114", false) +} + +func TestIPv4DestinationSessionsRetainInterfaces(t *testing.T) { + ifNames := []string{os.Getenv("FLUX_TEST_MULTICAST_TX"), os.Getenv("FLUX_TEST_MULTICAST_TX2")} + if ifNames[0] == "" || ifNames[1] == "" { + t.Skip("requires two isolated IPv4 multicast interfaces") + } + for _, network := range []string{"udp4", "udp"} { + for _, destination := range []string{"239.255.0.114", "224.0.0.114", "255.255.255.255"} { + t.Run(network+"/"+destination, func(t *testing.T) { + bind := net.IPv4zero + if network == "udp" { + bind = net.IPv6unspecified + } + server, err := net.ListenUDP(network, &net.UDPAddr{IP: bind}) + if err != nil { + t.Fatal(err) + } + defer server.Close() + pc, err := PreserveDestination(server) + if err != nil { + t.Fatal(err) + } + client, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("198.18.0.1")}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + raw, err := client.SyscallConn() + if err != nil { + t.Fatal(err) + } + var sockErr error + if err := raw.Control(func(fd uintptr) { + sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_BROADCAST, 1) + }); err != nil { + t.Fatal(err) + } + if sockErr != nil { + t.Fatal(sockErr) + } + target := &net.UDPAddr{IP: net.ParseIP(destination), Port: server.LocalAddr().(*net.UDPAddr).Port} + for _, name := range ifNames { + iface, err := net.InterfaceByName(name) + if err != nil { + t.Fatal(err) + } + if target.IP.IsMulticast() { + if err := ipv4.NewPacketConn(server).JoinGroup(iface, target); err != nil { + t.Fatal(err) + } + } + if _, err := ipv4.NewPacketConn(client).WriteTo([]byte(name), &ipv4.ControlMessage{ + Src: client.LocalAddr().(*net.UDPAddr).IP, IfIndex: iface.Index, + }, target); err != nil { + t.Fatal(err) + } + } + server.SetReadDeadline(time.Now().Add(3 * time.Second)) + ln := NewListener(pc, &ListenConfig{ + Backlog: 2, ReadQueueSize: 2, ReadBufferSize: 128, TTL: 10 * time.Second, Keepalive: true, + Logger: xlogger.NewLogger(xlogger.OutputOption(io.Discard)), + }) + defer ln.Close() + seen := make(map[string]bool) + for range ifNames { + c, err := ln.Accept() + if err != nil { + t.Fatalf("interface did not get a separate session: %v", err) + } + defer c.Close() + var buf [128]byte + n, err := c.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + name := string(buf[:n]) + if (name != ifNames[0] && name != ifNames[1]) || seen[name] || + c.LocalAddr().String() != target.String() || c.RemoteAddr().String() != client.LocalAddr().String() { + t.Fatalf("unexpected interface session: data=%q local=%s peer=%s", buf[:n], c.LocalAddr(), c.RemoteAddr()) + } + iface, err := net.InterfaceByName(name) + if err != nil { + t.Fatal(err) + } + messages, err := unix.ParseSocketControlMessage(c.(*conn).packet.control) + if err != nil || len(messages) != 1 { + t.Fatalf("invalid reply control: %v", err) + } + var info unix.Inet4Pktinfo + if _, err := binary.Decode(messages[0].Data, binary.NativeEndian, &info); err != nil { + t.Fatal(err) + } + if int(info.Ifindex) != iface.Index { + t.Fatalf("reply interface = %d, want %d", info.Ifindex, iface.Index) + } + seen[name] = true + } + }) + } + } +} + +func TestInterfaceLocalMulticastSessionsRetainZones(t *testing.T) { + testScopedMulticastSessions(t, "ff01::114", true) +} + +func testScopedMulticastSessions(t *testing.T, destination string, interfaceLocal bool) { + t.Helper() + rxNames := []string{os.Getenv("FLUX_TEST_MULTICAST_RX"), os.Getenv("FLUX_TEST_MULTICAST_RX2")} + txNames := []string{os.Getenv("FLUX_TEST_MULTICAST_TX"), os.Getenv("FLUX_TEST_MULTICAST_TX2")} + for i := range rxNames { + if rxNames[i] == "" || txNames[i] == "" { + t.Skip("requires two isolated multicast veth pairs") + } + } + if interfaceLocal { + // Interface-local multicast is looped back on the sending interface. + rxNames = txNames + } + group := net.ParseIP(destination) + server, err := net.ListenUDP("udp6", &net.UDPAddr{IP: net.IPv6unspecified}) + if err != nil { + t.Fatal(err) + } + defer server.Close() + clientIP := net.ParseIP("fd00:1::1") + client, err := net.ListenUDP("udp6", &net.UDPAddr{IP: clientIP}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + pc, err := PreserveDestination(server) + if err != nil { + t.Fatal(err) + } + for i, rxName := range rxNames { + rx, err := net.InterfaceByName(rxName) + if err != nil { + t.Fatal(err) + } + if err := ipv6.NewPacketConn(server).JoinGroup(rx, &net.UDPAddr{IP: group}); err != nil { + t.Fatal(err) + } + tx, err := net.InterfaceByName(txNames[i]) + if err != nil { + t.Fatal(err) + } + if err := ipv6.NewPacketConn(client).SetMulticastInterface(tx); err != nil { + t.Fatal(err) + } + if _, err := client.WriteToUDP([]byte(rxName), &net.UDPAddr{ + IP: group, Port: server.LocalAddr().(*net.UDPAddr).Port, Zone: txNames[i], + }); err != nil { + t.Fatal(err) + } + } + server.SetReadDeadline(time.Now().Add(3 * time.Second)) + ln := NewListener(pc, &ListenConfig{ + Backlog: 2, ReadQueueSize: 2, ReadBufferSize: 128, TTL: 10 * time.Second, Keepalive: true, + Logger: xlogger.NewLogger(xlogger.OutputOption(io.Discard)), + }) + defer ln.Close() + first, err := ln.Accept() + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := ln.Accept() + if err != nil { + t.Fatalf("second interface did not get a separate session: %v", err) + } + defer second.Close() + if first == second { + t.Fatal("interface-scoped multicast sessions were shared") + } + seen := make(map[string]bool) + for _, c := range []net.Conn{first, second} { + var buf [128]byte + n, err := c.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + zone := string(buf[:n]) + local, peer := c.LocalAddr().(*net.UDPAddr), c.RemoteAddr().(*net.UDPAddr) + if (zone != rxNames[0] && zone != rxNames[1]) || seen[zone] || !local.IP.Equal(group) || + local.Port != server.LocalAddr().(*net.UDPAddr).Port || local.Zone != zone || + !peer.IP.Equal(clientIP) || peer.Port != client.LocalAddr().(*net.UDPAddr).Port || peer.Zone != "" { + t.Fatalf("unexpected scoped session: data=%q local=%s peer=%s", buf[:n], local, peer) + } + var reply ipv6.ControlMessage + if err := reply.Parse(c.(*conn).packet.control); err != nil { + t.Fatal(err) + } + iface, err := net.InterfaceByName(zone) + if err != nil { + t.Fatal(err) + } + if reply.IfIndex != iface.Index { + t.Fatalf("reply interface = %d, want %d", reply.IfIndex, iface.Index) + } + seen[zone] = true + } +} diff --git a/go-gost/x/internal/net/udp/packet_other.go b/go-gost/x/internal/net/udp/packet_other.go new file mode 100644 index 0000000..f0d3ffe --- /dev/null +++ b/go-gost/x/internal/net/udp/packet_other.go @@ -0,0 +1,9 @@ +//go:build !linux + +package udp + +import "net" + +func PreserveDestination(c *net.UDPConn) (net.PacketConn, error) { + return c, nil +} diff --git a/go-gost/x/listener/udp/destination_linux_test.go b/go-gost/x/listener/udp/destination_linux_test.go new file mode 100644 index 0000000..ecd89d9 --- /dev/null +++ b/go-gost/x/listener/udp/destination_linux_test.go @@ -0,0 +1,289 @@ +package udp + +import ( + "context" + "fmt" + "io" + "net" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "golang.org/x/sys/unix" + + corelimiter "github.com/go-gost/core/limiter" + coretraffic "github.com/go-gost/core/limiter/traffic" + corelistener "github.com/go-gost/core/listener" + corestats "github.com/go-gost/core/observer/stats" + xadmission "github.com/go-gost/x/admission" + xtraffic "github.com/go-gost/x/limiter/traffic" + xlogger "github.com/go-gost/x/logger" + xmetadata "github.com/go-gost/x/metadata" + "github.com/go-gost/x/metrics" + xstats "github.com/go-gost/x/observer/stats" +) + +type countedTraffic struct { + coretraffic.TrafficLimiter + in, out atomic.Int64 +} + +func (l *countedTraffic) In(ctx context.Context, key string, opts ...corelimiter.Option) coretraffic.Limiter { + l.in.Add(1) + return l.TrafficLimiter.In(ctx, key, opts...) +} + +func (l *countedTraffic) Out(ctx context.Context, key string, opts ...corelimiter.Option) coretraffic.Limiter { + l.out.Add(1) + return l.TrafficLimiter.Out(ctx, key, opts...) +} + +func TestDestinationPreservedThroughWrappers(t *testing.T) { + v4 := []string{"127.0.0.1", "127.0.0.2"} + if address := os.Getenv("FLUX_TEST_IPV4_LINK_LOCAL"); address != "" { + v4 = append(v4, address) + } + v6 := strings.FieldsFunc(os.Getenv("FLUX_TEST_IPV6_ADDRS"), func(r rune) bool { return r == ',' }) + if len(v6) == 0 { + v6 = []string{"::1"} + } + for _, tc := range []struct { + name, bind string + targets []string + clientBind string + }{ + {"ipv4-wildcard", "0.0.0.0:0", v4, ""}, + {"dual-stack-wildcard", "[::]:0", append(v4, v6...), ""}, + {"fixed-ipv6-source", "[::]:0", v6, "::1"}, + {"explicit-ipv4", "127.0.0.1:0", []string{"127.0.0.1"}, ""}, + {"explicit-ipv6", "[::1]:0", []string{"::1"}, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + metrics.Enable(true) + defer metrics.Enable(false) + log := xlogger.NewLogger(xlogger.OutputOption(io.Discard)) + stats := xstats.NewStats(false) + lim := &countedTraffic{TrafficLimiter: xtraffic.NewTrafficLimiter( + xtraffic.LimitsOption("$ 1MB 1MB", "$$ 1MB 1MB"), xtraffic.LoggerOption(log), + )} + ln := NewListener(corelistener.AddrOption(tc.bind), corelistener.LoggerOption(log), + corelistener.ServiceOption(t.Name()), corelistener.StatsOption(stats), + corelistener.TrafficLimiterOption(lim), + corelistener.AdmissionOption(xadmission.NewAdmission( + xadmission.WhitelistOption(true), + xadmission.MatchersOption([]string{"127.0.0.0/8", "169.254.0.0/16", "::1", "fd00::/8"}), + xadmission.LoggerOption(log), + )), + ) + if err := ln.Init(xmetadata.NewMetadata(map[string]any{"keepAlive": true})); err != nil { + t.Fatal(err) + } + defer ln.Close() + clientIP := net.IPv6unspecified + if tc.clientBind != "" { + clientIP = net.ParseIP(tc.clientBind) + } + client, err := net.ListenUDP("udp", &net.UDPAddr{IP: clientIP}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + client.SetDeadline(time.Now().Add(5 * time.Second)) + clientAddr := client.LocalAddr().(*net.UDPAddr) + var accepted atomic.Int64 + errors := make(chan error, 128) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + accepted.Add(1) + go func() { + defer c.Close() + buf := make([]byte, 128) + for { + n, err := c.Read(buf) + if err != nil { + return + } + var round, target int + if _, err := fmt.Sscanf(string(buf[:n]), "%d/%d", &round, &target); err != nil { + errors <- err + return + } + local := c.LocalAddr().(*net.UDPAddr) + if local.IP.To4() != nil && local.Zone != "" { + errors <- fmt.Errorf("IPv4 destination has a zone: %s", local) + return + } + if !local.IP.Equal(net.ParseIP(tc.targets[target])) { + errors <- fmt.Errorf("local destination = %s, want %s", local.IP, tc.targets[target]) + return + } + peer, ok := c.RemoteAddr().(*net.UDPAddr) + if !ok { + errors <- fmt.Errorf("peer type changed: %T", c.RemoteAddr()) + return + } + if tc.clientBind != "" && (!peer.IP.Equal(clientAddr.IP) || peer.Port != clientAddr.Port) { + errors <- fmt.Errorf("peer endpoint = %s, want %s", peer, clientAddr) + return + } + if _, err := c.Write(buf[:n]); err != nil { + errors <- err + return + } + } + }() + } + }() + port := ln.Addr().(*net.UDPAddr).Port + var bytes int + for round := 0; round < 3; round++ { + for target, ip := range tc.targets { + payload := []byte(fmt.Sprintf("%d/%d", round, target)) + bytes += len(payload) + if _, err := client.WriteToUDP(payload, &net.UDPAddr{IP: net.ParseIP(ip), Port: port}); err != nil { + t.Fatal(err) + } + } + seen := make(map[int]bool) + for range tc.targets { + buf := make([]byte, 128) + n, from, err := client.ReadFromUDP(buf) + if err != nil { + select { + case serverErr := <-errors: + t.Fatal(serverErr) + default: + t.Fatal(err) + } + } + var gotRound, target int + if _, err := fmt.Sscanf(string(buf[:n]), "%d/%d", &gotRound, &target); err != nil { + t.Fatal(err) + } + if gotRound != round || seen[target] || from.Port != port || !from.IP.Equal(net.ParseIP(tc.targets[target])) { + t.Fatalf("unexpected reply %q from %s", buf[:n], from) + } + seen[target] = true + } + } + if accepted.Load() != int64(len(tc.targets)) { + t.Fatalf("accepted %d sessions, want %d", accepted.Load(), len(tc.targets)) + } + deadline := time.Now().Add(time.Second) + for stats.Get(corestats.KindOutputBytes) < uint64(bytes) && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if stats.Get(corestats.KindInputBytes) != uint64(bytes) || stats.Get(corestats.KindOutputBytes) != uint64(bytes) { + t.Fatalf("traffic stats lost: in=%d out=%d want=%d", stats.Get(corestats.KindInputBytes), stats.Get(corestats.KindOutputBytes), bytes) + } + if lim.in.Load() == 0 || lim.out.Load() == 0 { + t.Fatal("traffic limiter bypassed") + } + }) + } +} + +func TestBroadcastDestinationRepliesFromUnicast(t *testing.T) { + for _, bind := range []string{"0.0.0.0:0", "[::]:0"} { + t.Run(bind, func(t *testing.T) { + ln := NewListener(corelistener.AddrOption(bind), + corelistener.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard)))) + if err := ln.Init(xmetadata.NewMetadata(map[string]any{"keepAlive": true})); err != nil { + t.Fatal(err) + } + defer ln.Close() + client, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + raw, err := client.SyscallConn() + if err != nil { + t.Fatal(err) + } + var sockErr error + if err := raw.Control(func(fd uintptr) { + sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_BROADCAST, 1) + }); err != nil { + t.Fatal(err) + } + if sockErr != nil { + t.Fatal(sockErr) + } + client.SetDeadline(time.Now().Add(5 * time.Second)) + target := &net.UDPAddr{IP: net.IPv4(127, 255, 255, 255), Port: ln.Addr().(*net.UDPAddr).Port} + if _, err := client.WriteToUDP([]byte("broadcast"), target); err != nil { + t.Fatal(err) + } + conn, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if conn.LocalAddr().String() != target.String() { + t.Fatalf("local destination = %s, want %s", conn.LocalAddr(), target) + } + if conn.RemoteAddr().String() != client.LocalAddr().String() { + t.Fatalf("peer endpoint = %s, want %s", conn.RemoteAddr(), client.LocalAddr()) + } + var buf [128]byte + n, err := conn.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + if _, err := conn.Write(buf[:n]); err != nil { + t.Fatalf("reply to broadcast request: %v", err) + } + n, from, err := client.ReadFromUDP(buf[:]) + if err != nil { + t.Fatal(err) + } + if !from.IP.Equal(net.IPv4(127, 0, 0, 1)) || from.Port != target.Port || string(buf[:n]) != "broadcast" { + t.Fatalf("unexpected broadcast reply %q from %s", buf[:n], from) + } + }) + } +} + +func TestDestinationSessionExpires(t *testing.T) { + ln := NewListener(corelistener.AddrOption("0.0.0.0:0"), + corelistener.LoggerOption(xlogger.NewLogger(xlogger.OutputOption(io.Discard)))) + if err := ln.Init(xmetadata.NewMetadata(map[string]any{"keepAlive": true, "ttl": "20ms"})); err != nil { + t.Fatal(err) + } + defer ln.Close() + client, err := net.DialUDP("udp4", nil, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 2), Port: ln.Addr().(*net.UDPAddr).Port}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + client.Write([]byte("first")) + first, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatal(err) + } + var b [16]byte + first.Read(b[:]) + time.Sleep(80 * time.Millisecond) + if _, err := first.Read(b[:]); err == nil { + t.Fatal("idle session did not expire") + } + client.Write([]byte("second")) + second, err := acceptWithTimeout(t, ln, time.Second) + if err != nil { + t.Fatal(err) + } + defer second.Close() + if first == second { + t.Fatal("expired session reused") + } + if second.LocalAddr().(*net.UDPAddr).IP.String() != "127.0.0.2" { + t.Fatalf("new session lost destination: %s", second.LocalAddr()) + } +} diff --git a/go-gost/x/listener/udp/listener.go b/go-gost/x/listener/udp/listener.go index 3ac3119..56cc695 100644 --- a/go-gost/x/listener/udp/listener.go +++ b/go-gost/x/listener/udp/listener.go @@ -56,11 +56,15 @@ func (l *udpListener) Init(md md.Metadata) (err error) { return } - var conn net.PacketConn - conn, err = net.ListenUDP(network, laddr) + socket, err := net.ListenUDP(network, laddr) if err != nil { return } + conn, err := udp.PreserveDestination(socket) + if err != nil { + socket.Close() + return err + } conn = metrics.WrapPacketConn(l.options.Service, conn) conn = stats.WrapPacketConn(conn, l.options.Stats) conn = admission.WrapPacketConn(l.options.Admission, conn)