ping: Fix stale flows kept alive by unrelated ICMP traffic

Unconnected raw ICMP sockets receive every ICMP packet arriving at the
host, so any ICMP traffic refreshed the read deadline of every flow and
stale flows (with their raw sockets and goroutines) were only reclaimed
by LRU eviction while processing all host ICMP traffic in the meantime.

Expire flows based on their own activity only, and on Linux attach a
classic BPF ident filter to each raw socket so other flows' packets are
dropped in the kernel instead of waking every flow.

ping/destination.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/sagernet/sing-tun"
 	"github.com/sagernet/sing-tun/internal/gtcpip/header"
+	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/buf"
 	"github.com/sagernet/sing/common/control"
 	E "github.com/sagernet/sing/common/exceptions"
@@ -29,6 +30,7 @@ type Destination struct {
 	destination   netip.Addr
 	routeContext  tun.DirectRouteContext
 	timeout       time.Duration
+	lastActive    common.TypedValue[time.Time]
 	requestAccess sync.Mutex
 	requests      map[pingRequest]time.Time
 }
@@ -73,6 +75,7 @@ func ConnectDestination(
 		timeout:      timeout,
 		requests:     make(map[pingRequest]time.Time),
 	}
+	d.lastActive.Store(time.Now())
 	go d.loopRead()
 	return d, nil
 }
@@ -80,14 +83,21 @@ func ConnectDestination(
 func (d *Destination) loopRead() {
 	defer d.Close()
 	for {
-		buffer := buf.NewSize(maxICMPPacketSize)
+		deadline := d.lastActive.Load().Add(d.timeout)
-		err := d.conn.SetReadDeadline(time.Now().Add(d.timeout))
+		if !time.Now().Before(deadline) {
+			return
+		}
+		err := d.conn.SetReadDeadline(deadline)
 		if err != nil {
 			d.logger.ErrorContext(d.ctx, E.Cause(err, "set read deadline for ICMP conn"))
 		}
+		buffer := buf.NewSize(maxICMPPacketSize)
 		err = d.conn.ReadIP(buffer)
 		if err != nil {
 			buffer.Release()
+			if E.IsTimeout(err) {
+				continue
+			}
 			if !E.IsClosed(err) {
 				d.logger.ErrorContext(d.ctx, E.Cause(err, "receive ICMP echo reply"))
 			}
@@ -158,6 +168,7 @@ func (d *Destination) loopRead() {
 			}
 			d.logger.TraceContext(d.ctx, "read ICMPv6 echo reply from ", ipHdr.SourceAddr(), " to ", ipHdr.DestinationAddr(), " id ", icmpHdr.Ident(), " seq ", icmpHdr.Sequence())
 		}
+		d.lastActive.Store(time.Now())
 		err = d.routeContext.WritePacket(buffer.Bytes())
 		if err != nil {
 			d.logger.ErrorContext(d.ctx, E.Cause(err, "write ICMP echo reply"))
@@ -167,6 +178,7 @@ func (d *Destination) loopRead() {
 }
 
 func (d *Destination) WritePacket(packet *buf.Buffer) error {
+	d.lastActive.Store(time.Now())
 	if !d.destination.Is6() {
 		ipHdr := header.IPv4(packet.Bytes())
 		if !ipHdr.IsValid(packet.Len()) {

ping/destination_gvisor.go

@@ -23,12 +23,13 @@ import (
 var _ tun.DirectRouteDestination = (*GVisorDestination)(nil)
 
 type GVisorDestination struct {
-	ctx      context.Context
+	ctx        context.Context
-	logger   logger.ContextLogger
+	logger     logger.ContextLogger
-	endpoint tcpip.Endpoint
+	endpoint   tcpip.Endpoint
-	conn     *gonet.TCPConn
+	conn       *gonet.TCPConn
-	rewriter *SourceRewriter
+	rewriter   *SourceRewriter
-	timeout  time.Duration
+	timeout    time.Duration
+	lastActive common.TypedValue[time.Time]
 }
 
 func ConnectGVisor(
@@ -86,6 +87,7 @@ func ConnectGVisor(
 		rewriter: rewriter,
 		timeout:  timeout,
 	}
+	destination.lastActive.Store(time.Now())
 	go destination.loopRead()
 	return destination, nil
 }
@@ -93,29 +95,41 @@ func ConnectGVisor(
 func (d *GVisorDestination) loopRead() {
 	defer d.endpoint.Close()
 	for {
-		buffer := buf.NewSize(maxICMPPacketSize)
+		deadline := d.lastActive.Load().Add(d.timeout)
-		err := d.conn.SetReadDeadline(time.Now().Add(d.timeout))
+		if !time.Now().Before(deadline) {
+			return
+		}
+		err := d.conn.SetReadDeadline(deadline)
 		if err != nil {
 			d.logger.ErrorContext(d.ctx, E.Cause(err, "set read deadline for ICMP conn"))
 		}
+		buffer := buf.NewSize(maxICMPPacketSize)
 		n, err := d.conn.Read(buffer.FreeBytes())
 		if err != nil {
 			buffer.Release()
+			if E.IsTimeout(err) {
+				continue
+			}
 			if !E.IsClosed(err) {
 				d.logger.ErrorContext(d.ctx, E.Cause(err, "receive ICMP echo reply"))
 			}
 			return
 		}
 		buffer.Truncate(n)
-		_, err = d.rewriter.WriteBack(buffer.Bytes())
+		var matched bool
+		matched, err = d.rewriter.WriteBack(buffer.Bytes())
 		if err != nil {
 			d.logger.ErrorContext(d.ctx, E.Cause(err, "write ICMP echo reply"))
 		}
+		if matched {
+			d.lastActive.Store(time.Now())
+		}
 		buffer.Release()
 	}
 }
 
 func (d *GVisorDestination) WritePacket(packet *buf.Buffer) error {
+	d.lastActive.Store(time.Now())
 	d.rewriter.RewritePacket(packet.Bytes())
 	return common.Error(d.conn.Write(packet.Bytes()))
 }

ping/destination_test.go

@@ -2,11 +2,16 @@ package ping_test
 
 import (
 	"context"
+	"errors"
 	"net/netip"
+	"os"
+	"slices"
 	"testing"
 	"time"
 
+	"github.com/sagernet/sing-tun/internal/gtcpip/header"
 	"github.com/sagernet/sing-tun/ping"
+	"github.com/sagernet/sing/common/buf"
 	"github.com/sagernet/sing/common/logger"
 
 	"github.com/stretchr/testify/require"
@@ -22,3 +27,83 @@ func TestIsClosed(t *testing.T) {
 	destination.Close()
 	require.True(t, destination.IsClosed())
 }
+
+type channelWriter struct {
+	packets chan []byte
+}
+
+func (w *channelWriter) WritePacket(packet []byte) error {
+	select {
+	case w.packets <- slices.Clone(packet):
+	default:
+	}
+	return nil
+}
+
+type discardWriter struct{}
+
+func (w discardWriter) WritePacket(packet []byte) error {
+	return nil
+}
+
+func buildEchoRequest(source, destination netip.Addr, identifier, sequence uint16) *buf.Buffer {
+	const totalLen = header.IPv4MinimumSize + header.ICMPv4MinimumSize
+	packet := buf.NewSize(totalLen)
+	ipHdr := header.IPv4(packet.Extend(totalLen))
+	ipHdr.Encode(&header.IPv4Fields{
+		TotalLength: totalLen,
+		TTL:         64,
+		Protocol:    uint8(header.ICMPv4ProtocolNumber),
+		SrcAddr:     source,
+		DstAddr:     destination,
+	})
+	ipHdr.SetChecksum(^ipHdr.CalculateChecksum())
+	icmpHdr := header.ICMPv4(ipHdr.Payload())
+	icmpHdr.SetType(header.ICMPv4Echo)
+	icmpHdr.SetIdent(identifier)
+	icmpHdr.SetSequence(sequence)
+	icmpHdr.SetChecksum(header.ICMPv4Checksum(icmpHdr, 0))
+	return packet
+}
+
+// darwin unprivileged ICMP sockets and Linux raw sockets receive every
+// loopback ICMP packet regardless of the flow it belongs to, so this test
+// cannot run parallel to TestPing.
+func TestDestinationIdleExpiry(t *testing.T) {
+	loopback := netip.MustParseAddr("127.0.0.1")
+	writer := &channelWriter{packets: make(chan []byte, 16)}
+	destination, err := ping.ConnectDestination(context.Background(), logger.NOP(), nil, loopback, writer, time.Second)
+	if errors.Is(err, os.ErrPermission) {
+		t.SkipNow()
+	}
+	require.NoError(t, err)
+	defer destination.Close()
+
+	err = destination.WritePacket(buildEchoRequest(loopback, loopback, 0x1111, 1))
+	require.NoError(t, err)
+	select {
+	case packet := <-writer.packets:
+		replyIPHdr := header.IPv4(packet)
+		replyICMPHdr := header.ICMPv4(replyIPHdr.Payload())
+		require.Equal(t, header.ICMPv4EchoReply, replyICMPHdr.Type())
+		require.Equal(t, uint16(0x1111), replyICMPHdr.Ident())
+	case <-time.After(3 * time.Second):
+		t.Fatal("no echo reply received")
+	}
+
+	noise, err := ping.ConnectDestination(context.Background(), logger.NOP(), nil, loopback, discardWriter{}, 30*time.Second)
+	require.NoError(t, err)
+	defer noise.Close()
+
+	deadline := time.Now().Add(5 * time.Second)
+	var sequence uint16
+	for time.Now().Before(deadline) {
+		sequence++
+		_ = noise.WritePacket(buildEchoRequest(loopback, loopback, 0x2222, sequence))
+		if destination.IsClosed() {
+			return
+		}
+		time.Sleep(150 * time.Millisecond)
+	}
+	t.Fatal("flow not closed after idle timeout despite unrelated ICMP traffic")
+}

ping/filter_linux.go

@@ -0,0 +1,99 @@
+package ping
+
+import (
+	"sync"
+	"syscall"
+
+	"github.com/sagernet/sing-tun/internal/gtcpip/header"
+	"github.com/sagernet/sing/common"
+	"github.com/sagernet/sing/common/control"
+
+	"golang.org/x/net/bpf"
+	"golang.org/x/sys/unix"
+)
+
+type identFilterState struct {
+	access     sync.Mutex
+	attached   bool
+	disabled   bool
+	identifier uint16
+}
+
+// The kernel clones every matching-protocol packet into every unconnected raw
+// ICMP socket, so without a socket filter each flow receives and discards all
+// other flows' traffic.
+func (c *Conn) updateIdentFilter(wireIdentifier uint16) {
+	if !c.privileged {
+		return
+	}
+	syscallConn, isSyscallConn := common.Cast[syscall.Conn](c.conn)
+	if !isSyscallConn {
+		return
+	}
+	state := &c.identFilter
+	state.access.Lock()
+	defer state.access.Unlock()
+	if state.disabled {
+		return
+	}
+	if state.attached {
+		if state.identifier == wireIdentifier {
+			return
+		}
+		state.attached = false
+		state.disabled = true
+		_ = control.Conn(syscallConn, func(fd uintptr) error {
+			return unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_DETACH_FILTER, 0)
+		})
+		return
+	}
+	rawInstructions, err := identFilterProgram(c.destination.Is6(), wireIdentifier)
+	if err != nil {
+		state.disabled = true
+		return
+	}
+	filter := make([]unix.SockFilter, len(rawInstructions))
+	for i, instruction := range rawInstructions {
+		filter[i] = unix.SockFilter{Code: instruction.Op, Jt: instruction.Jt, Jf: instruction.Jf, K: instruction.K}
+	}
+	program := unix.SockFprog{Len: uint16(len(filter)), Filter: &filter[0]}
+	err = control.Conn(syscallConn, func(fd uintptr) error {
+		return unix.SetsockoptSockFprog(int(fd), unix.SOL_SOCKET, unix.SO_ATTACH_FILTER, &program)
+	})
+	if err != nil {
+		state.disabled = true
+		return
+	}
+	state.attached = true
+	state.identifier = wireIdentifier
+}
+
+func identFilterProgram(is6 bool, wireIdentifier uint16) ([]bpf.RawInstruction, error) {
+	if !is6 {
+		// Raw ICMPv4 sockets deliver the full packet including the IP header.
+		return bpf.Assemble([]bpf.Instruction{
+			bpf.LoadMemShift{Off: 0},
+			bpf.LoadIndirect{Off: 0, Size: 1},
+			bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(header.ICMPv4EchoReply), SkipTrue: 3},
+			bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(header.ICMPv4DstUnreachable), SkipTrue: 4},
+			bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(header.ICMPv4TimeExceeded), SkipTrue: 3},
+			bpf.RetConstant{Val: 0},
+			bpf.LoadIndirect{Off: 4, Size: 2},
+			bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(wireIdentifier), SkipFalse: 1},
+			bpf.RetConstant{Val: 0xffffffff},
+			bpf.RetConstant{Val: 0},
+		})
+	}
+	// Raw ICMPv6 sockets deliver the ICMPv6 message without the IP header.
+	return bpf.Assemble([]bpf.Instruction{
+		bpf.LoadAbsolute{Off: 0, Size: 1},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(header.ICMPv6EchoReply), SkipTrue: 3},
+		bpf.JumpIf{Cond: bpf.JumpGreaterThan, Val: uint32(header.ICMPv6ParamProblem), SkipTrue: 1},
+		bpf.RetConstant{Val: 0xffffffff},
+		bpf.RetConstant{Val: 0},
+		bpf.LoadAbsolute{Off: 4, Size: 2},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(wireIdentifier), SkipFalse: 1},
+		bpf.RetConstant{Val: 0xffffffff},
+		bpf.RetConstant{Val: 0},
+	})
+}

ping/filter_other.go

@@ -0,0 +1,8 @@
+//go:build !linux
+
+package ping
+
+type identFilterState struct{}
+
+func (c *Conn) updateIdentFilter(wireIdentifier uint16) {
+}

ping/ping.go

@@ -28,6 +28,7 @@ type Conn struct {
 	destination netip.Addr
 	source      common.TypedValue[netip.Addr]
 	closed      atomic.Bool
+	identFilter identFilterState
 	readMsg     func(b, oob []byte) (n, oobn int, addr netip.Addr, err error)
 }
 
@@ -268,6 +269,7 @@ func (c *Conn) WriteIP(buffer *buf.Buffer) error {
 			icmpHdr := header.ICMPv4(ipHdr.Payload())
 			icmpHdr.SetIdent(^icmpHdr.Ident())
 			icmpHdr.SetChecksum(header.ICMPv4Checksum(icmpHdr, 0))
+			c.updateIdentFilter(icmpHdr.Ident())
 		}
 		c.source.Store(M.AddrFromIP(ipHdr.SourceAddressSlice()))
 		return common.Error(c.conn.Write(ipHdr.Payload()))
@@ -285,6 +287,7 @@ func (c *Conn) WriteIP(buffer *buf.Buffer) error {
 				Src:    ipHdr.SourceAddressSlice(),
 				Dst:    ipHdr.DestinationAddressSlice(),
 			}))
+			c.updateIdentFilter(icmpHdr.Ident())
 		}
 		c.source.Store(M.AddrFromIP(ipHdr.SourceAddressSlice()))
 		return common.Error(c.conn.Write(ipHdr.Payload()))

ping/ping_test.go

@@ -250,11 +250,6 @@ func testPingIPv4WriteIP(t *testing.T, privileged bool, addr string) {
 	defer response.Release()
 	err = conn.ReadIP(response)
 	require.NoError(t, err)
-	if runtime.GOOS == "linux" && privileged {
-		response.Reset()
-		err = conn.ReadIP(response)
-		require.NoError(t, err)
-	}
 	respIP := header.IPv4(response.Bytes())
 	require.NotZero(t, respIP.TTL())
 	respICMP := header.ICMPv4(respIP.Payload())
@@ -299,7 +294,7 @@ func testPingIPv6WriteIP(t *testing.T, privileged bool, addr string) {
 	defer response.Release()
 	err = conn.ReadIP(response)
 	require.NoError(t, err)
-	if runtime.GOOS == "darwin" || runtime.GOOS == "linux" && privileged {
+	if runtime.GOOS == "darwin" {
 		response.Reset()
 		err = conn.ReadIP(response)
 		require.NoError(t, err)