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)