From 30e535973b1d91488b73d32659e027e0071f61e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Thu, 9 Jul 2026 22:07:14 +0800 Subject: [PATCH] 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 | 16 +++++- ping/destination_gvisor.go | 32 ++++++++---- ping/destination_test.go | 85 ++++++++++++++++++++++++++++++++ ping/filter_linux.go | 99 ++++++++++++++++++++++++++++++++++++++ ping/filter_other.go | 8 +++ ping/ping.go | 3 ++ ping/ping_test.go | 7 +-- 7 files changed, 233 insertions(+), 17 deletions(-) create mode 100644 ping/filter_linux.go create mode 100644 ping/filter_other.go diff --git a/ping/destination.go b/ping/destination.go index e16763b..d43730c 100644 --- a/ping/destination.go +++ b/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) - err := d.conn.SetReadDeadline(time.Now().Add(d.timeout)) + deadline := d.lastActive.Load().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()) { diff --git a/ping/destination_gvisor.go b/ping/destination_gvisor.go index 25fd36e..0c508ee 100644 --- a/ping/destination_gvisor.go +++ b/ping/destination_gvisor.go @@ -23,12 +23,13 @@ import ( var _ tun.DirectRouteDestination = (*GVisorDestination)(nil) type GVisorDestination struct { - ctx context.Context - logger logger.ContextLogger - endpoint tcpip.Endpoint - conn *gonet.TCPConn - rewriter *SourceRewriter - timeout time.Duration + ctx context.Context + logger logger.ContextLogger + endpoint tcpip.Endpoint + conn *gonet.TCPConn + rewriter *SourceRewriter + 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) - err := d.conn.SetReadDeadline(time.Now().Add(d.timeout)) + deadline := d.lastActive.Load().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())) } diff --git a/ping/destination_test.go b/ping/destination_test.go index d0a1af8..db71139 100644 --- a/ping/destination_test.go +++ b/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") +} diff --git a/ping/filter_linux.go b/ping/filter_linux.go new file mode 100644 index 0000000..13f32a3 --- /dev/null +++ b/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}, + }) +} diff --git a/ping/filter_other.go b/ping/filter_other.go new file mode 100644 index 0000000..34c2137 --- /dev/null +++ b/ping/filter_other.go @@ -0,0 +1,8 @@ +//go:build !linux + +package ping + +type identFilterState struct{} + +func (c *Conn) updateIdentFilter(wireIdentifier uint16) { +} diff --git a/ping/ping.go b/ping/ping.go index 81ac696..23bf907 100644 --- a/ping/ping.go +++ b/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())) diff --git a/ping/ping_test.go b/ping/ping_test.go index 9fef5bf..be5f6a7 100644 --- a/ping/ping_test.go +++ b/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)