Add L3 forwarding support

transport/wireguard/device.go

@@ -5,7 +5,6 @@ import (
 	"net/netip"
 	"time"
 
-	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-tun"
 	"github.com/sagernet/sing/common/logger"
 	N "github.com/sagernet/sing/common/network"
@@ -45,8 +44,3 @@ func NewDevice(options DeviceOptions) (Device, error) {
 		return newSystemStackDevice(options)
 	}
 }
-
-type NatDevice interface {
-	Device
-	CreateDestination(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error)
-}

transport/wireguard/device_nat.go

@@ -1,103 +0,0 @@
-package wireguard
-
-import (
-	"context"
-	"sync/atomic"
-	"time"
-
-	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/log"
-	"github.com/sagernet/sing-tun"
-	"github.com/sagernet/sing-tun/ping"
-	"github.com/sagernet/sing/common/buf"
-	"github.com/sagernet/sing/common/logger"
-)
-
-var _ Device = (*natDeviceWrapper)(nil)
-
-type natDeviceWrapper struct {
-	Device
-	ctx            context.Context
-	logger         logger.ContextLogger
-	packetOutbound chan *buf.Buffer
-	rewriter       *ping.SourceRewriter
-	buffer         [][]byte
-}
-
-func NewNATDevice(ctx context.Context, logger logger.ContextLogger, upstream Device) NatDevice {
-	wrapper := &natDeviceWrapper{
-		Device:         upstream,
-		ctx:            ctx,
-		logger:         logger,
-		packetOutbound: make(chan *buf.Buffer, 256),
-		rewriter:       ping.NewSourceRewriter(ctx, logger, upstream.Inet4Address(), upstream.Inet6Address()),
-	}
-	return wrapper
-}
-
-func (d *natDeviceWrapper) Read(bufs [][]byte, sizes []int, offset int) (n int, err error) {
-	select {
-	case packet := <-d.packetOutbound:
-		defer packet.Release()
-		sizes[0] = copy(bufs[0][offset:], packet.Bytes())
-		return 1, nil
-	default:
-	}
-	return d.Device.Read(bufs, sizes, offset)
-}
-
-func (d *natDeviceWrapper) Write(bufs [][]byte, offset int) (int, error) {
-	for _, buffer := range bufs {
-		handled, err := d.rewriter.WriteBack(buffer[offset:])
-		if handled {
-			if err != nil {
-				return 0, err
-			}
-		} else {
-			d.buffer = append(d.buffer, buffer)
-		}
-	}
-	if len(d.buffer) > 0 {
-		_, err := d.Device.Write(d.buffer, offset)
-		if err != nil {
-			return 0, err
-		}
-		d.buffer = d.buffer[:0]
-	}
-	return 0, nil
-}
-
-func (d *natDeviceWrapper) CreateDestination(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error) {
-	ctx := log.ContextWithNewID(d.ctx)
-	session := tun.DirectRouteSession{
-		Source:      metadata.Source.Addr,
-		Destination: metadata.Destination.Addr,
-	}
-	d.rewriter.CreateSession(session, routeContext)
-	d.logger.InfoContext(ctx, "linked ", metadata.Network, " connection from ", metadata.Source.AddrString(), " to ", metadata.Destination.AddrString())
-	return &natDestination{device: d, session: session}, nil
-}
-
-var _ tun.DirectRouteDestination = (*natDestination)(nil)
-
-type natDestination struct {
-	device  *natDeviceWrapper
-	session tun.DirectRouteSession
-	closed  atomic.Bool
-}
-
-func (d *natDestination) WritePacket(buffer *buf.Buffer) error {
-	d.device.rewriter.RewritePacket(buffer.Bytes())
-	d.device.packetOutbound <- buffer
-	return nil
-}
-
-func (d *natDestination) Close() error {
-	d.closed.Store(true)
-	d.device.rewriter.DeleteSession(d.session)
-	return nil
-}
-
-func (d *natDestination) IsClosed() bool {
-	return d.closed.Load()
-}

transport/wireguard/device_stack.go

@@ -8,7 +8,6 @@ import (
 	"net/netip"
 	"os"
 	"sync"
-	"time"
 
 	"github.com/sagernet/gvisor/pkg/buffer"
 	"github.com/sagernet/gvisor/pkg/tcpip"
@@ -20,10 +19,7 @@ import (
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/icmp"
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/tcp"
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/udp"
-	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-tun"
-	"github.com/sagernet/sing-tun/ping"
 	"github.com/sagernet/sing/common/buf"
 	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
@@ -32,11 +28,9 @@ import (
 	wgTun "github.com/sagernet/wireguard-go/tun"
 )
 
-var _ NatDevice = (*stackDevice)(nil)
+var _ Device = (*stackDevice)(nil)
 
 type stackDevice struct {
-	ctx            context.Context
-	logger         log.ContextLogger
 	stack          *stack.Stack
 	mtu            uint32
 	events         chan wgTun.Event
@@ -47,12 +41,11 @@ type stackDevice struct {
 	dispatcher     stack.NetworkDispatcher
 	inet4Address   netip.Addr
 	inet6Address   netip.Addr
+	icmpForwarder  *tun.ICMPForwarder
 }
 
 func newStackDevice(options DeviceOptions) (*stackDevice, error) {
 	tunDevice := &stackDevice{
-		ctx:            options.Context,
-		logger:         options.Logger,
 		mtu:            options.MTU,
 		events:         make(chan wgTun.Event, 1),
 		outbound:       make(chan *stack.PacketBuffer, 256),
@@ -63,10 +56,6 @@ func newStackDevice(options DeviceOptions) (*stackDevice, error) {
 	if err != nil {
 		return nil, err
 	}
-	var (
-		inet4Address netip.Addr
-		inet6Address netip.Addr
-	)
 	for _, prefix := range options.Address {
 		addr := tun.AddressFromAddr(prefix.Addr())
 		protoAddr := tcpip.ProtocolAddress{
@@ -76,12 +65,10 @@ func newStackDevice(options DeviceOptions) (*stackDevice, error) {
 			},
 		}
 		if prefix.Addr().Is4() {
-			inet4Address = prefix.Addr()
-			tunDevice.inet4Address = inet4Address
+			tunDevice.inet4Address = prefix.Addr()
 			protoAddr.Protocol = ipv4.ProtocolNumber
 		} else {
-			inet6Address = prefix.Addr()
-			tunDevice.inet6Address = inet6Address
+			tunDevice.inet6Address = prefix.Addr()
 			protoAddr.Protocol = ipv6.ProtocolNumber
 		}
 		gErr := ipStack.AddProtocolAddress(tun.DefaultNIC, protoAddr, stack.AddressProperties{})
@@ -93,10 +80,10 @@ func newStackDevice(options DeviceOptions) (*stackDevice, error) {
 	if options.Handler != nil {
 		ipStack.SetTransportProtocolHandler(tcp.ProtocolNumber, tun.NewTCPForwarder(options.Context, ipStack, options.Handler).HandlePacket)
 		ipStack.SetTransportProtocolHandler(udp.ProtocolNumber, tun.NewUDPForwarder(options.Context, ipStack, options.Handler, options.UDPTimeout).HandlePacket)
-		icmpForwarder := tun.NewICMPForwarder(options.Context, ipStack, options.Logger, options.Handler, options.ICMPTimeout)
-		icmpForwarder.SetLocalAddresses(inet4Address, inet6Address)
+		icmpForwarder := tun.NewICMPForwarder(ipStack, options.Handler, options.Logger)
 		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber4, icmpForwarder.HandlePacket)
 		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber6, icmpForwarder.HandlePacket)
+		tunDevice.icmpForwarder = icmpForwarder
 	}
 	return tunDevice, nil
 }
@@ -255,6 +242,9 @@ func (w *stackDevice) Close() error {
 	w.closeOnce.Do(func() {
 		close(w.done)
 		close(w.events)
+		if w.icmpForwarder != nil {
+			w.icmpForwarder.Close()
+		}
 		w.stack.Close()
 		for _, endpoint := range w.stack.CleanupEndpoints() {
 			endpoint.Abort()
@@ -268,23 +258,6 @@ func (w *stackDevice) BatchSize() int {
 	return 1
 }
 
-func (w *stackDevice) CreateDestination(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error) {
-	ctx := log.ContextWithNewID(w.ctx)
-	destination, err := ping.ConnectGVisor(
-		ctx, w.logger,
-		metadata.Source.Addr, metadata.Destination.Addr,
-		routeContext,
-		w.stack,
-		w.inet4Address, w.inet6Address,
-		timeout,
-	)
-	if err != nil {
-		return nil, err
-	}
-	w.logger.InfoContext(ctx, "linked ", metadata.Network, " connection from ", metadata.Source.AddrString(), " to ", metadata.Destination.AddrString())
-	return destination, nil
-}
-
 var _ stack.LinkEndpoint = (*wireEndpoint)(nil)
 
 type wireEndpoint stackDevice

transport/wireguard/device_system_stack.go

@@ -3,10 +3,8 @@
 package wireguard
 
 import (
-	"context"
 	"net/netip"
 	"sync"
-	"time"
 
 	"github.com/sagernet/gvisor/pkg/buffer"
 	"github.com/sagernet/gvisor/pkg/tcpip"
@@ -17,12 +15,8 @@ import (
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/icmp"
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/tcp"
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/udp"
-	"github.com/sagernet/sing-box/adapter"
-	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-tun"
-	"github.com/sagernet/sing-tun/ping"
 	E "github.com/sagernet/sing/common/exceptions"
-	"github.com/sagernet/sing/common/logger"
 	"github.com/sagernet/wireguard-go/device"
 )
 
@@ -30,12 +24,11 @@ var _ Device = (*systemStackDevice)(nil)
 
 type systemStackDevice struct {
 	*systemDevice
-	ctx       context.Context
-	logger    logger.ContextLogger
-	stack     *stack.Stack
-	endpoint  *deviceEndpoint
-	writeBufs [][]byte
-	closeOnce sync.Once
+	stack         *stack.Stack
+	endpoint      *deviceEndpoint
+	icmpForwarder *tun.ICMPForwarder
+	writeBufs     [][]byte
+	closeOnce     sync.Once
 }
 
 func newSystemStackDevice(options DeviceOptions) (*systemStackDevice, error) {
@@ -51,10 +44,6 @@ func newSystemStackDevice(options DeviceOptions) (*systemStackDevice, error) {
 	if err != nil {
 		return nil, err
 	}
-	var (
-		inet4Address netip.Addr
-		inet6Address netip.Addr
-	)
 	for _, prefix := range options.Address {
 		addr := tun.AddressFromAddr(prefix.Addr())
 		protoAddr := tcpip.ProtocolAddress{
@@ -64,10 +53,8 @@ func newSystemStackDevice(options DeviceOptions) (*systemStackDevice, error) {
 			},
 		}
 		if prefix.Addr().Is4() {
-			inet4Address = prefix.Addr()
 			protoAddr.Protocol = ipv4.ProtocolNumber
 		} else {
-			inet6Address = prefix.Addr()
 			protoAddr.Protocol = ipv6.ProtocolNumber
 		}
 		gErr := ipStack.AddProtocolAddress(tun.DefaultNIC, protoAddr, stack.AddressProperties{})
@@ -75,21 +62,20 @@ func newSystemStackDevice(options DeviceOptions) (*systemStackDevice, error) {
 			return nil, E.New("parse local address ", protoAddr.AddressWithPrefix, ": ", gErr.String())
 		}
 	}
-	if options.Handler != nil {
-		ipStack.SetTransportProtocolHandler(tcp.ProtocolNumber, tun.NewTCPForwarder(options.Context, ipStack, options.Handler).HandlePacket)
-		ipStack.SetTransportProtocolHandler(udp.ProtocolNumber, tun.NewUDPForwarder(options.Context, ipStack, options.Handler, options.UDPTimeout).HandlePacket)
-		icmpForwarder := tun.NewICMPForwarder(options.Context, ipStack, options.Logger, options.Handler, options.ICMPTimeout)
-		icmpForwarder.SetLocalAddresses(inet4Address, inet6Address)
-		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber4, icmpForwarder.HandlePacket)
-		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber6, icmpForwarder.HandlePacket)
-	}
-	return &systemStackDevice{
-		ctx:          options.Context,
-		logger:       options.Logger,
+	stackDevice := &systemStackDevice{
 		systemDevice: system,
 		stack:        ipStack,
 		endpoint:     endpoint,
-	}, nil
+	}
+	if options.Handler != nil {
+		ipStack.SetTransportProtocolHandler(tcp.ProtocolNumber, tun.NewTCPForwarder(options.Context, ipStack, options.Handler).HandlePacket)
+		ipStack.SetTransportProtocolHandler(udp.ProtocolNumber, tun.NewUDPForwarder(options.Context, ipStack, options.Handler, options.UDPTimeout).HandlePacket)
+		icmpForwarder := tun.NewICMPForwarder(ipStack, options.Handler, options.Logger)
+		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber4, icmpForwarder.HandlePacket)
+		ipStack.SetTransportProtocolHandler(icmp.ProtocolNumber6, icmpForwarder.HandlePacket)
+		stackDevice.icmpForwarder = icmpForwarder
+	}
+	return stackDevice, nil
 }
 
 func (w *systemStackDevice) SetDevice(device *device.Device) {
@@ -129,6 +115,9 @@ func (w *systemStackDevice) Close() error {
 	var err error
 	w.closeOnce.Do(func() {
 		close(w.endpoint.done)
+		if w.icmpForwarder != nil {
+			w.icmpForwarder.Close()
+		}
 		w.stack.Close()
 		for _, endpoint := range w.stack.CleanupEndpoints() {
 			endpoint.Abort()
@@ -165,23 +154,6 @@ func (w *systemStackDevice) writeStack(packet []byte) bool {
 	return true
 }
 
-func (w *systemStackDevice) CreateDestination(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error) {
-	ctx := log.ContextWithNewID(w.ctx)
-	destination, err := ping.ConnectGVisor(
-		ctx, w.logger,
-		metadata.Source.Addr, metadata.Destination.Addr,
-		routeContext,
-		w.stack,
-		w.inet4Address, w.inet6Address,
-		timeout,
-	)
-	if err != nil {
-		return nil, err
-	}
-	w.logger.InfoContext(ctx, "linked ", metadata.Network, " connection from ", metadata.Source.AddrString(), " to ", metadata.Destination.AddrString())
-	return destination, nil
-}
-
 type deviceEndpoint struct {
 	mtu        uint32
 	done       chan struct{}

transport/wireguard/endpoint.go

@@ -10,12 +10,9 @@ import (
 	"os"
 	"reflect"
 	"strings"
-	"time"
 	"unsafe"
 
-	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/dialer"
-	"github.com/sagernet/sing-tun"
 	"github.com/sagernet/sing/common"
 	E "github.com/sagernet/sing/common/exceptions"
 	F "github.com/sagernet/sing/common/format"
@@ -35,7 +32,7 @@ type Endpoint struct {
 	ipcConf        string
 	allowedAddress []netip.Prefix
 	tunDevice      Device
-	natDevice      NatDevice
+	returnDevice   *returnDeviceWrapper
 	device         *device.Device
 	allowedIPs     *device.AllowedIPs
 	pause          pause.Manager
@@ -120,17 +117,13 @@ func NewEndpoint(options EndpointOptions) (*Endpoint, error) {
 	if err != nil {
 		return nil, E.Cause(err, "create WireGuard device")
 	}
-	natDevice, isNatDevice := tunDevice.(NatDevice)
-	if !isNatDevice {
-		natDevice = NewNATDevice(options.Context, options.Logger, tunDevice)
-	}
 	return &Endpoint{
 		options:        options,
 		peers:          peers,
 		ipcConf:        ipcConf,
 		allowedAddress: allowedAddresses,
 		tunDevice:      tunDevice,
-		natDevice:      natDevice,
+		returnDevice:   &returnDeviceWrapper{Device: tunDevice},
 	}, nil
 }
 
@@ -157,7 +150,11 @@ func (e *Endpoint) Start(resolve bool) error {
 	var bind conn.Bind
 	wgListener, isWgListener := common.Cast[dialer.WireGuardListener](e.options.Dialer)
 	if isWgListener {
-		bind = conn.NewStdNetBind(wgListener.WireGuardControl())
+		stdBind := conn.NewStdNetBind(wgListener.WireGuardControl())
+		if e.options.ListenPort == 0 && len(e.peers) == 1 && e.peers[0].endpoint.IsValid() {
+			stdBind.(*conn.StdNetBind).SetSinglePeerMode()
+		}
+		bind = stdBind
 	} else {
 		var (
 			isConnect   bool
@@ -190,13 +187,7 @@ func (e *Endpoint) Start(resolve bool) error {
 			e.options.Logger.Error(fmt.Sprintf(strings.ToLower(format), args...))
 		},
 	}
-	var deviceInput Device
-	if e.natDevice != nil {
-		deviceInput = e.natDevice
-	} else {
-		deviceInput = e.tunDevice
-	}
-	wgDevice := device.NewDevice(e.options.Context, deviceInput, bind, logger, e.options.Workers)
+	wgDevice := device.NewDevice(e.options.Context, e.returnDevice, bind, logger, e.options.Workers)
 	e.tunDevice.SetDevice(wgDevice)
 	var ipcConf strings.Builder
 	ipcConf.WriteString(e.ipcConf)
@@ -251,13 +242,6 @@ func (e *Endpoint) Lookup(address netip.Addr) *device.Peer {
 	return e.allowedIPs.Lookup(address.AsSlice())
 }
 
-func (e *Endpoint) NewDirectRouteConnection(metadata adapter.InboundContext, routeContext tun.DirectRouteContext, timeout time.Duration) (tun.DirectRouteDestination, error) {
-	if e.natDevice == nil {
-		return nil, os.ErrInvalid
-	}
-	return e.natDevice.CreateDestination(metadata, routeContext, timeout)
-}
-
 func (e *Endpoint) onPauseUpdated(event int) {
 	switch event {
 	case pause.EventDevicePaused, pause.EventNetworkPause:

transport/wireguard/port.go

@@ -0,0 +1,157 @@
+package wireguard
+
+import (
+	"net/netip"
+	"sync/atomic"
+
+	"github.com/sagernet/sing-tun"
+	"github.com/sagernet/sing-tun/gtcpip/header"
+	E "github.com/sagernet/sing/common/exceptions"
+	"github.com/sagernet/wireguard-go/device"
+)
+
+func (e *Endpoint) PortAddresses() (netip.Addr, netip.Addr) {
+	return e.tunDevice.Inet4Address(), e.tunDevice.Inet6Address()
+}
+
+func (e *Endpoint) PortMTU() uint32 {
+	return e.options.MTU
+}
+
+func (e *Endpoint) WritePackets(packets [][]byte) error {
+	wgDevice := e.device
+	if wgDevice == nil {
+		return E.New("WireGuard device is not ready")
+	}
+	packetRefs := make([]*device.InputPacketRef, 0, len(packets))
+	refs := make([]device.InputPacketRef, len(packets))
+	packetSlices := make([][]byte, len(packets))
+	for i, packet := range packets {
+		if len(packet) == 0 {
+			continue
+		}
+		var destination []byte
+		switch header.IPVersion(packet) {
+		case header.IPv4Version:
+			if len(packet) < header.IPv4MinimumSize {
+				continue
+			}
+			destination = header.IPv4(packet).DestinationAddressSlice()
+		case header.IPv6Version:
+			if len(packet) < header.IPv6MinimumSize {
+				continue
+			}
+			destination = header.IPv6(packet).DestinationAddressSlice()
+		default:
+			continue
+		}
+		packetSlices[i] = packet
+		refs[i] = device.InputPacketRef{
+			Destination:  destination,
+			PacketSlices: packetSlices[i : i+1],
+		}
+		packetRefs = append(packetRefs, &refs[i])
+	}
+	if len(packetRefs) == 0 {
+		return nil
+	}
+	unmatchedRefs := wgDevice.InputPackets(packetRefs)
+	if len(unmatchedRefs) == 0 {
+		return nil
+	}
+	state := e.returnDevice.state.Load()
+	if state == nil {
+		return nil
+	}
+	var replies [][]byte
+	for _, packetRef := range unmatchedRefs {
+		packet := packetRef.PacketSlices[0]
+		var source netip.Addr
+		if header.IPVersion(packet) == header.IPv4Version {
+			source = e.tunDevice.Inet4Address()
+		} else {
+			source = e.tunDevice.Inet6Address()
+		}
+		reply, replyOk := tun.BuildUnreachable(packet, source, state.headroom)
+		if replyOk {
+			replies = append(replies, reply)
+		}
+	}
+	if len(replies) > 0 {
+		state.returnPath.ReturnPackets(replies)
+	}
+	return nil
+}
+
+func (e *Endpoint) AttachReturn(returnPath tun.Return) error {
+	headroom := returnPath.ReturnHeadroom()
+	if headroom > device.MessageTransportOffsetContent {
+		return E.New("return path headroom ", headroom, " exceeds available ", device.MessageTransportOffsetContent)
+	}
+	newState := &returnPathState{
+		returnPath: returnPath,
+		headroom:   headroom,
+	}
+	for {
+		currentState := e.returnDevice.state.Load()
+		if currentState != nil {
+			if currentState.returnPath == returnPath {
+				return nil
+			}
+			return E.New("return path already attached")
+		}
+		if e.returnDevice.state.CompareAndSwap(nil, newState) {
+			return nil
+		}
+	}
+}
+
+func (e *Endpoint) DetachReturn(returnPath tun.Return) error {
+	currentState := e.returnDevice.state.Load()
+	if currentState != nil && currentState.returnPath == returnPath {
+		e.returnDevice.state.CompareAndSwap(currentState, nil)
+	}
+	return nil
+}
+
+type returnPathState struct {
+	returnPath tun.Return
+	headroom   int
+}
+
+type returnDeviceWrapper struct {
+	Device
+	state atomic.Pointer[returnPathState]
+}
+
+func (d *returnDeviceWrapper) Write(bufs [][]byte, offset int) (int, error) {
+	state := d.state.Load()
+	if state == nil || len(bufs) == 0 {
+		return d.Device.Write(bufs, offset)
+	}
+	packets := make([][]byte, len(bufs))
+	for i, packet := range bufs {
+		// wireguard-go leaves device.MessageTransportOffsetContent writable bytes in front of the decrypted packet.
+		packets[i] = packet[offset-state.headroom:]
+	}
+	unconsumed := state.returnPath.ReturnPackets(packets)
+	if len(unconsumed) == 0 {
+		return 0, nil
+	}
+	if len(unconsumed) == len(bufs) {
+		return d.Device.Write(bufs, offset)
+	}
+	remaining := make([][]byte, 0, len(unconsumed))
+	searchIndex := 0
+	for _, packet := range unconsumed {
+		for searchIndex < len(bufs) && &packet[0] != &bufs[searchIndex][offset-state.headroom] {
+			searchIndex++
+		}
+		if searchIndex == len(bufs) {
+			break
+		}
+		remaining = append(remaining, bufs[searchIndex])
+		searchIndex++
+	}
+	return d.Device.Write(remaining, offset)
+}