From 03423427b5b94ba40fecfdcde77b9e02669ac19e Mon Sep 17 00:00:00 2001 From: lubingtan Date: Wed, 5 Aug 2026 20:58:54 +0800 Subject: [PATCH] atunnel: support IPv6 original destination lookup --- internal/atunnel/original_dst_linux.go | 68 +++- internal/atunnel/original_dst_linux_test.go | 395 ++++++++++++++++++++ 2 files changed, 444 insertions(+), 19 deletions(-) create mode 100644 internal/atunnel/original_dst_linux_test.go diff --git a/internal/atunnel/original_dst_linux.go b/internal/atunnel/original_dst_linux.go index 07dd0f934..8b4309119 100644 --- a/internal/atunnel/original_dst_linux.go +++ b/internal/atunnel/original_dst_linux.go @@ -18,6 +18,7 @@ package atunnel import ( "encoding/binary" + "errors" "fmt" "net" "strconv" @@ -26,10 +27,12 @@ import ( "golang.org/x/sys/unix" ) -// TCPOriginalDestination reads the IPv4 destination preserved by a Linux -// REDIRECT rule. Actor networking is currently IPv4-only. -// TODO(liorlieberman) add the IPv6 IP6T_SO_ORIGINAL_DST variant -// when actor veth setup gains dual-stack support. +// IP6T_SO_ORIGINAL_DST is not generated by golang.org/x/sys/unix. It is +// defined as 80 in linux/netfilter_ipv6/ip6_tables.h. +const ip6tSOOriginalDst = 80 + +// TCPOriginalDestination reads the IPv4 or IPv6 destination preserved by a +// Linux REDIRECT rule. func TCPOriginalDestination(conn net.Conn) (string, error) { tcpConn, ok := conn.(*net.TCPConn) if !ok { @@ -40,21 +43,15 @@ func TCPOriginalDestination(conn net.Conn) (string, error) { return "", fmt.Errorf("atunnel: acquiring TCP syscall connection: %w", err) } - var addr unix.RawSockaddrInet4 var sockoptErr error + var destination string if err := rawConn.Control(func(fd uintptr) { - size := uint32(unsafe.Sizeof(addr)) - _, _, errno := unix.Syscall6( - unix.SYS_GETSOCKOPT, - fd, - unix.SOL_IP, - unix.SO_ORIGINAL_DST, - uintptr(unsafe.Pointer(&addr)), - uintptr(unsafe.Pointer(&size)), - 0, - ) - if errno != 0 { - sockoptErr = errno + destination, sockoptErr = originalIPv4Destination(fd) + // Linux returns ENOENT when the IPv4 original-destination option is + // queried on a redirected IPv6 connection. Only then try the IPv6 + // equivalent, so unrelated IPv4 failures retain their original error. + if errors.Is(sockoptErr, unix.ENOENT) { + destination, sockoptErr = originalIPv6Destination(fd) } }); err != nil { return "", fmt.Errorf("atunnel: accessing TCP socket: %w", err) @@ -62,11 +59,44 @@ func TCPOriginalDestination(conn net.Conn) (string, error) { if sockoptErr != nil { return "", fmt.Errorf("atunnel: reading original TCP destination: %w", sockoptErr) } + return destination, nil +} + +func originalIPv4Destination(fd uintptr) (string, error) { + var addr unix.RawSockaddrInet4 + if errno := getOriginalDestination(fd, unix.SOL_IP, unix.SO_ORIGINAL_DST, unsafe.Pointer(&addr), unsafe.Sizeof(addr)); errno != 0 { + return "", errno + } + return formatOriginalDestination(addr.Addr[:], addr.Port) +} + +func originalIPv6Destination(fd uintptr) (string, error) { + var addr unix.RawSockaddrInet6 + if errno := getOriginalDestination(fd, unix.SOL_IPV6, ip6tSOOriginalDst, unsafe.Pointer(&addr), unsafe.Sizeof(addr)); errno != 0 { + return "", errno + } + return formatOriginalDestination(addr.Addr[:], addr.Port) +} + +func getOriginalDestination(fd uintptr, level, option int, addr unsafe.Pointer, addrSize uintptr) unix.Errno { + size := uint32(addrSize) + _, _, errno := unix.Syscall6( + unix.SYS_GETSOCKOPT, + fd, + uintptr(level), + uintptr(option), + uintptr(addr), + uintptr(unsafe.Pointer(&size)), + 0, + ) + return errno +} - portBytes := (*[2]byte)(unsafe.Pointer(&addr.Port)) +func formatOriginalDestination(ip []byte, rawPort uint16) (string, error) { + portBytes := (*[2]byte)(unsafe.Pointer(&rawPort)) port := binary.BigEndian.Uint16(portBytes[:]) if port == 0 { return "", fmt.Errorf("atunnel: original TCP destination has port zero") } - return net.JoinHostPort(net.IP(addr.Addr[:]).String(), strconv.Itoa(int(port))), nil + return net.JoinHostPort(net.IP(ip).String(), strconv.Itoa(int(port))), nil } diff --git a/internal/atunnel/original_dst_linux_test.go b/internal/atunnel/original_dst_linux_test.go new file mode 100644 index 000000000..4f26d9805 --- /dev/null +++ b/internal/atunnel/original_dst_linux_test.go @@ -0,0 +1,395 @@ +//go:build linux + +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package atunnel + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "strings" + "testing" + "time" + + "github.com/google/nftables" + "github.com/google/nftables/binaryutil" + "github.com/google/nftables/expr" + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" + "golang.org/x/sys/unix" + + "github.com/agent-substrate/substrate/internal/ateomnet" + "github.com/agent-substrate/substrate/internal/roottest" +) + +func TestTCPOriginalDestination(t *testing.T) { + roottest.Require(t, "CAP_NET_ADMIN + CAP_SYS_ADMIN for an actor-like network namespace and nftables REDIRECT rule") + + // Model the production path rather than redirecting a locally generated + // connection through OUTPUT. Actor egress enters the worker netns through a + // veth and is redirected in PREROUTING; that is the path on which Linux + // preserves SO_ORIGINAL_DST for atunnel. + actorNS := newTestNetNS(t) + actorIP, hostIP := setupTestVeth(t, actorNS) + // targetListener reserves the port the actor intends to reach. The NAT rule + // below must prevent connections from reaching it. + // + // redirectListener represents atunnel's local egress listener. It receives + // the redirected connection and is therefore the connection on which we ask + // Linux for the original destination. + redirectListener := listenTCP(t, hostIP) + defer redirectListener.Close() + targetListener := listenTCP(t, hostIP) + defer targetListener.Close() + targetPort := targetListener.Addr().(*net.TCPAddr).Port + + table := &nftables.Table{Family: nftables.TableFamilyIPv4, Name: fmt.Sprintf("atunnel_original_dst_test_%d", os.Getpid())} + installOriginalDstRedirect(t, table, actorIP, targetPort, redirectListener.Addr().(*net.TCPAddr).Port) + + clientDone := make(chan error, 1) + go func() { + // From the actor's perspective this is an ordinary connection to + // hostIP:targetPort. The worker's PREROUTING rule redirects it before + // it reaches the host network stack's local delivery path. + clientDone <- ateomnet.NetNSDo(context.Background(), actorNS, func(context.Context) error { + conn, err := net.DialTimeout("tcp4", net.JoinHostPort(hostIP.String(), fmt.Sprint(targetPort)), time.Second) + if err == nil { + _ = conn.Close() + } + return err + }) + }() + + if err := redirectListener.(*net.TCPListener).SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + redirected, err := redirectListener.Accept() + if err != nil { + t.Fatalf("accepting redirected connection: %v", err) + } + defer redirected.Close() + + // The accepted socket is addressed to redirectListener, but the kernel's + // SO_ORIGINAL_DST record must still contain the destination chosen by the + // actor before nftables rewrote it. + got, err := TCPOriginalDestination(redirected) + if err != nil { + t.Fatalf("TCPOriginalDestination: %v", err) + } + want := net.JoinHostPort(hostIP.String(), fmt.Sprint(targetPort)) + if got != want { + t.Errorf("original destination = %q, want %q", got, want) + } + if err := <-clientDone; err != nil { + t.Fatalf("dialing redirected connection: %v", err) + } +} + +func TestTCPOriginalDestinationIPv6(t *testing.T) { + roottest.Require(t, "CAP_NET_ADMIN + CAP_SYS_ADMIN for an actor-like network namespace and nftables REDIRECT rule") + + actorNS := newTestNetNS(t) + actorIP, hostIP := setupTestIPv6Veth(t, actorNS) + redirectListener := listenTCP6(t, hostIP) + defer redirectListener.Close() + targetListener := listenTCP6(t, hostIP) + defer targetListener.Close() + targetPort := targetListener.Addr().(*net.TCPAddr).Port + + table := &nftables.Table{Family: nftables.TableFamilyIPv6, Name: fmt.Sprintf("atunnel_original_dst_ipv6_test_%d", os.Getpid())} + installOriginalDstIPv6Redirect(t, table, actorIP, targetPort, redirectListener.Addr().(*net.TCPAddr).Port) + + clientDone := make(chan error, 1) + go func() { + clientDone <- ateomnet.NetNSDo(context.Background(), actorNS, func(context.Context) error { + conn, err := net.DialTimeout("tcp6", net.JoinHostPort(hostIP.String(), fmt.Sprint(targetPort)), time.Second) + if err == nil { + _ = conn.Close() + } + return err + }) + }() + + if err := redirectListener.(*net.TCPListener).SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + redirected, err := redirectListener.Accept() + if err != nil { + t.Fatalf("accepting redirected IPv6 connection: %v", err) + } + defer redirected.Close() + + // This assertion captures the IPv6 behavior required by #686. + got, err := TCPOriginalDestination(redirected) + if err != nil { + t.Fatalf("TCPOriginalDestination: %v", err) + } + want := net.JoinHostPort(hostIP.String(), fmt.Sprint(targetPort)) + if got != want { + t.Errorf("original IPv6 destination = %q, want %q", got, want) + } + if err := <-clientDone; err != nil { + t.Fatalf("dialing redirected IPv6 connection: %v", err) + } +} + +func newTestNetNS(t *testing.T) netns.NsHandle { + t.Helper() + name := fmt.Sprintf("atunnel-original-dst-%d", os.Getpid()) + ns, err := ateomnet.CreateNetNSWithoutSwitching(name) + if err != nil { + if errors.Is(err, unix.EPERM) || strings.Contains(err.Error(), "operation not permitted") { + t.Skipf("needs CAP_SYS_ADMIN to create network namespace: %v", err) + } + t.Fatal(err) + } + t.Cleanup(func() { + _ = ns.Close() + if err := netns.DeleteNamed(name); err != nil { + t.Errorf("deleting test network namespace: %v", err) + } + }) + return ns +} + +func setupTestVeth(t *testing.T, actorNS netns.NsHandle) (actorIP, hostIP net.IP) { + t.Helper() + hostName := fmt.Sprintf("atod%d", os.Getpid()) + peerName := fmt.Sprintf("atop%d", os.Getpid()) + if err := netlink.LinkAdd(&netlink.Veth{LinkAttrs: netlink.LinkAttrs{Name: hostName}, PeerName: peerName}); err != nil { + if errors.Is(err, unix.EPERM) || strings.Contains(err.Error(), "operation not permitted") { + t.Skipf("needs CAP_NET_ADMIN to create veth: %v", err) + } + t.Fatal(err) + } + t.Cleanup(func() { + if link, err := netlink.LinkByName(hostName); err == nil { + if err := netlink.LinkDel(link); err != nil { + t.Errorf("deleting test veth: %v", err) + } + } + }) + hostLink, err := netlink.LinkByName(hostName) + if err != nil { + t.Fatal(err) + } + // Allocate one of the /30s in 198.18.0.0/16 from the PID so concurrent + // test processes do not try to use the same host-side address. + network := uint16(os.Getpid() % (1 << 14)) + thirdOctet := byte(network >> 6) + fourthOctet := byte(network&0x3f) << 2 + hostIP = net.IPv4(198, 18, thirdOctet, fourthOctet+1) + actorIP = net.IPv4(198, 18, thirdOctet, fourthOctet+2) + if err := netlink.AddrAdd(hostLink, &netlink.Addr{IPNet: &net.IPNet{IP: hostIP, Mask: net.CIDRMask(30, 32)}}); err != nil { + t.Fatal(err) + } + if err := netlink.LinkSetUp(hostLink); err != nil { + t.Fatal(err) + } + peer, err := netlink.LinkByName(peerName) + if err != nil { + t.Fatal(err) + } + if err := netlink.LinkSetNsFd(peer, int(actorNS)); err != nil { + t.Fatal(err) + } + // Complete the actor end of the point-to-point link inside its own netns. + if err := ateomnet.NetNSDo(context.Background(), actorNS, func(context.Context) error { + lo, err := netlink.LinkByName("lo") + if err != nil { + return err + } + if err := netlink.LinkSetUp(lo); err != nil { + return err + } + link, err := netlink.LinkByName(peerName) + if err != nil { + return err + } + if err := netlink.AddrAdd(link, &netlink.Addr{IPNet: &net.IPNet{IP: actorIP, Mask: net.CIDRMask(30, 32)}}); err != nil { + return err + } + return netlink.LinkSetUp(link) + }); err != nil { + t.Fatal(err) + } + return actorIP, hostIP +} + +func listenTCP(t *testing.T, hostIP net.IP) net.Listener { + t.Helper() + listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: hostIP, Port: 0}) + if err != nil { + t.Fatal(err) + } + return listener +} + +func setupTestIPv6Veth(t *testing.T, actorNS netns.NsHandle) (actorIP, hostIP net.IP) { + t.Helper() + hostName := fmt.Sprintf("atod6%d", os.Getpid()) + peerName := fmt.Sprintf("atop6%d", os.Getpid()) + if err := netlink.LinkAdd(&netlink.Veth{LinkAttrs: netlink.LinkAttrs{Name: hostName}, PeerName: peerName}); err != nil { + if errors.Is(err, unix.EPERM) || strings.Contains(err.Error(), "operation not permitted") { + t.Skipf("needs CAP_NET_ADMIN to create veth: %v", err) + } + t.Fatal(err) + } + t.Cleanup(func() { + if link, err := netlink.LinkByName(hostName); err == nil { + if err := netlink.LinkDel(link); err != nil { + t.Errorf("deleting test IPv6 veth: %v", err) + } + } + }) + hostLink, err := netlink.LinkByName(hostName) + if err != nil { + t.Fatal(err) + } + prefix := uint16(os.Getpid()) + hostIP = net.ParseIP(fmt.Sprintf("fd00:198:18:%x::1", prefix)) + actorIP = net.ParseIP(fmt.Sprintf("fd00:198:18:%x::2", prefix)) + // This isolated veth has no competing IPv6 peers. Suppress DAD so the + // address can be bound immediately instead of remaining tentative while + // the test is trying to start its listener. + if err := netlink.AddrAdd(hostLink, &netlink.Addr{IPNet: &net.IPNet{IP: hostIP, Mask: net.CIDRMask(64, 128)}, Flags: unix.IFA_F_NODAD}); err != nil { + t.Fatal(err) + } + if err := netlink.LinkSetUp(hostLink); err != nil { + t.Fatal(err) + } + peer, err := netlink.LinkByName(peerName) + if err != nil { + t.Fatal(err) + } + if err := netlink.LinkSetNsFd(peer, int(actorNS)); err != nil { + t.Fatal(err) + } + if err := ateomnet.NetNSDo(context.Background(), actorNS, func(context.Context) error { + lo, err := netlink.LinkByName("lo") + if err != nil { + return err + } + if err := netlink.LinkSetUp(lo); err != nil { + return err + } + link, err := netlink.LinkByName(peerName) + if err != nil { + return err + } + if err := netlink.AddrAdd(link, &netlink.Addr{IPNet: &net.IPNet{IP: actorIP, Mask: net.CIDRMask(64, 128)}, Flags: unix.IFA_F_NODAD}); err != nil { + return err + } + return netlink.LinkSetUp(link) + }); err != nil { + t.Fatal(err) + } + return actorIP, hostIP +} + +func listenTCP6(t *testing.T, hostIP net.IP) net.Listener { + t.Helper() + listener, err := net.ListenTCP("tcp6", &net.TCPAddr{IP: hostIP, Port: 0}) + if err != nil { + t.Fatal(err) + } + return listener +} + +func installOriginalDstRedirect(t *testing.T, table *nftables.Table, actorIP net.IP, targetPort, redirectPort int) { + t.Helper() + c := &nftables.Conn{} + c.AddTable(table) + chain := c.AddChain(&nftables.Chain{ + Name: "prerouting", + Table: table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPrerouting, + Priority: nftables.ChainPriorityNATDest, + }) + c.AddRule(&nftables.Rule{ + Table: table, + Chain: chain, + Exprs: []expr.Any{ + &expr.Meta{Key: expr.MetaKeyL4PROTO, Register: 1}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: []byte{unix.IPPROTO_TCP}}, + // Restrict the rule to this test's actor so the temporary table cannot + // affect unrelated local TCP traffic. + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseNetworkHeader, Offset: 12, Len: 4}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: actorIP.To4()}, + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseTransportHeader, Offset: 2, Len: 2}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: binaryutil.BigEndian.PutUint16(uint16(targetPort))}, + &expr.Immediate{Register: 1, Data: binaryutil.BigEndian.PutUint16(uint16(redirectPort))}, + &expr.Redir{RegisterProtoMin: 1}, + }, + }) + if err := c.Flush(); err != nil { + if errors.Is(err, unix.EPERM) || strings.Contains(err.Error(), "operation not permitted") { + t.Skipf("needs CAP_NET_ADMIN to install nftables rule: %v", err) + } + t.Fatalf("installing nftables redirect: %v", err) + } + t.Cleanup(func() { + cleanup := &nftables.Conn{} + cleanup.DelTable(table) + if err := cleanup.Flush(); err != nil { + t.Errorf("removing nftables redirect: %v", err) + } + }) +} + +func installOriginalDstIPv6Redirect(t *testing.T, table *nftables.Table, actorIP net.IP, targetPort, redirectPort int) { + t.Helper() + c := &nftables.Conn{} + c.AddTable(table) + chain := c.AddChain(&nftables.Chain{ + Name: "prerouting", + Table: table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPrerouting, + Priority: nftables.ChainPriorityNATDest, + }) + c.AddRule(&nftables.Rule{ + Table: table, + Chain: chain, + Exprs: []expr.Any{ + &expr.Meta{Key: expr.MetaKeyL4PROTO, Register: 1}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: []byte{unix.IPPROTO_TCP}}, + // An IPv6 source address begins eight bytes into the IPv6 header. + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseNetworkHeader, Offset: 8, Len: 16}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: actorIP.To16()}, + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseTransportHeader, Offset: 2, Len: 2}, + &expr.Cmp{Op: expr.CmpOpEq, Register: 1, Data: binaryutil.BigEndian.PutUint16(uint16(targetPort))}, + &expr.Immediate{Register: 1, Data: binaryutil.BigEndian.PutUint16(uint16(redirectPort))}, + &expr.Redir{RegisterProtoMin: 1}, + }, + }) + if err := c.Flush(); err != nil { + if errors.Is(err, unix.EPERM) || strings.Contains(err.Error(), "operation not permitted") { + t.Skipf("needs CAP_NET_ADMIN to install IPv6 nftables rule: %v", err) + } + t.Fatalf("installing IPv6 nftables redirect: %v", err) + } + t.Cleanup(func() { + cleanup := &nftables.Conn{} + cleanup.DelTable(table) + if err := cleanup.Flush(); err != nil { + t.Errorf("removing IPv6 nftables redirect: %v", err) + } + }) +}