diff --git a/common/interrupt/conn.go b/common/interrupt/conn.go index 94caa915d4..e56690bdcb 100644 --- a/common/interrupt/conn.go +++ b/common/interrupt/conn.go @@ -4,6 +4,7 @@ import ( "net" "github.com/sagernet/sing/common/bufio" + N "github.com/sagernet/sing/common/network" "github.com/sagernet/sing/common/x/list" ) @@ -52,6 +53,31 @@ type PacketConn struct { element *list.Element[*groupConnItem] } +type SingPacketConn struct { + N.PacketConn + group *Group + element *list.Element[*groupConnItem] +} + +func (c *SingPacketConn) Close() error { + c.group.access.Lock() + defer c.group.access.Unlock() + c.group.connections.Remove(c.element) + return c.PacketConn.Close() +} + +func (c *SingPacketConn) ReaderReplaceable() bool { + return true +} + +func (c *SingPacketConn) WriterReplaceable() bool { + return true +} + +func (c *SingPacketConn) Upstream() any { + return c.PacketConn +} + /*func (c *PacketConn) MarkAsInternal() { c.element.Value.internal = true }*/ diff --git a/common/interrupt/group.go b/common/interrupt/group.go index ba2e7f739b..bd3fbb0a2a 100644 --- a/common/interrupt/group.go +++ b/common/interrupt/group.go @@ -5,6 +5,7 @@ import ( "net" "sync" + N "github.com/sagernet/sing/common/network" "github.com/sagernet/sing/common/x/list" ) @@ -36,6 +37,13 @@ func (g *Group) NewPacketConn(conn net.PacketConn, isExternal bool) net.PacketCo return &PacketConn{PacketConn: conn, group: g, element: item} } +func (g *Group) NewSingPacketConn(conn N.PacketConn, isExternal bool) N.PacketConn { + g.access.Lock() + defer g.access.Unlock() + item := g.connections.PushBack(&groupConnItem{conn, isExternal}) + return &SingPacketConn{PacketConn: conn, group: g, element: item} +} + func (g *Group) Interrupt(interruptExternalConnections bool) { g.access.Lock() defer g.access.Unlock() diff --git a/common/windivert/integration_windows_test.go b/common/windivert/integration_windows_test.go index 100ee0d4b9..4020db8e20 100644 --- a/common/windivert/integration_windows_test.go +++ b/common/windivert/integration_windows_test.go @@ -8,7 +8,6 @@ import ( "net/netip" "os" "path/filepath" - "strconv" "testing" "time" @@ -110,36 +109,33 @@ func stopDriver(t *testing.T) { } // The image lock on the cached .sys can outlive SERVICE_STOPPED by tens of -// seconds (observed on GitHub-hosted runners), but it only blocks writes -// and deletes — rename is permitted. Move the locked file aside instead of -// waiting for the release. -func plantTamperedDriver(t *testing.T, target string, planted []byte) { +// seconds (observed on GitHub-hosted runners), and on current runner images +// it blocks renames as well as writes and deletes. Tests that need to tamper +// with the cache therefore redirect it to a directory the kernel has never +// loaded a driver from. Not t.TempDir: once StartService maps a .sys from +// the directory, the image lock makes the cleanup RemoveAll fail the test. +func setTempDriverCache(t *testing.T) { t.Helper() - require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o755)) - err := os.WriteFile(target, planted, 0o644) - if err == nil { - return - } - moved := target + ".locked-" + strconv.FormatInt(time.Now().UnixNano(), 10) - require.NoError(t, os.Rename(target, moved)) - t.Cleanup(func() { os.Remove(moved) }) - require.NoError(t, os.WriteFile(target, planted, 0o644)) + dir, err := os.MkdirTemp("", "sing-box-windivert-test-") + require.NoError(t, err) + t.Cleanup(func() { os.RemoveAll(dir) }) + t.Setenv("LocalAppData", dir) } // A foreign .sys planted in the user-writable cache must never reach // StartService: the install path has to detect the mismatch against the // embedded asset and repair the file before handing it to SCM. func TestIntegrationTamperedCacheRepaired(t *testing.T) { - // Open/close once so a working install is the baseline, then stop the - // driver so the cached file is writable for tampering. - h := openHandle(t, nil, FlagSendOnly) - require.NoError(t, h.Close()) + setTempDriverCache(t) + // The driver left running by earlier tests would satisfy Open without + // touching the cache; stop it so the install path runs. stopDriver(t) target := cachedDriverPath(t) - plantTamperedDriver(t, target, []byte("planted payload, not the WinDivert driver")) + require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o755)) + require.NoError(t, os.WriteFile(target, []byte("planted payload, not the WinDivert driver"), 0o644)) - h = openHandle(t, nil, FlagSendOnly) + h := openHandle(t, nil, FlagSendOnly) require.NoError(t, h.Close()) content, err := os.ReadFile(target) @@ -151,11 +147,10 @@ func TestIntegrationTamperedCacheRepaired(t *testing.T) { // install completes; without this, the file could be swapped between // verification and the kernel mapping it. func TestIntegrationDriverFileLockedWhileHeld(t *testing.T) { - target := cachedDriverPath(t) - // Stop the driver so the kernel image lock is gone and the failures - // asserted below can only come from the handle extractVerified holds. - stopDriver(t) - plantTamperedDriver(t, target, sysBytes) + // A fresh cache directory guarantees the failures asserted below can + // only come from the handle extractVerified holds, not a kernel image + // lock left by earlier tests. + setTempDriverCache(t) sysPath, sysFile, err := extractVerified() require.NoError(t, err) diff --git a/protocol/bridge/netfilter_linux.go b/protocol/bridge/netfilter_linux.go index 2d1e4cace1..38a8588b4e 100644 --- a/protocol/bridge/netfilter_linux.go +++ b/protocol/bridge/netfilter_linux.go @@ -29,33 +29,71 @@ var ( func enableBridgeForwarding(logger logger.ContextLogger, tunName string, inet4 bool, inet6 bool) []sysctlState { var restore []sysctlState - enable := func(path string) { + enable := func(path string) bool { content, err := os.ReadFile(path) if err != nil { logger.Debug(E.Cause(err, "read ", path)) - return + return false } value := strings.TrimSpace(string(content)) if value == "1" { - return + return false } err = os.WriteFile(path, []byte("1"), 0o644) if err != nil { logger.Debug(E.Cause(err, "enable ", path)) - return + return false } restore = append(restore, sysctlState{name: path, value: value}) + return true } if inet4 { enable("/proc/sys/net/ipv4/ip_forward") } if inet6 { - enable("/proc/sys/net/ipv6/conf/all/forwarding") + if enable("/proc/sys/net/ipv6/conf/all/forwarding") { + restore = append(restore, overruleAcceptRA(logger)...) + } } _ = os.WriteFile("/proc/sys/net/ipv4/conf/"+tunName+"/rp_filter", []byte("2"), 0o644) return restore } +// Writing conf/all/forwarding copies forwarding=1 to conf/default and to every +// existing interface (addrconf_fixup_forwarding), and ipv6_accept_ra() then +// requires accept_ra=2 on a forwarding interface; raise interfaces left at the +// host default of 1 so SLAAC (e.g. on PPPoE WANs) survives forwarding. +// conf/default is included so interfaces created afterwards inherit 2. +func overruleAcceptRA(logger logger.ContextLogger) []sysctlState { + var restore []sysctlState + entries, err := os.ReadDir("/proc/sys/net/ipv6/conf") + if err != nil { + logger.Debug(E.Cause(err, "read /proc/sys/net/ipv6/conf")) + return nil + } + for _, entry := range entries { + if entry.Name() == "all" { + continue + } + path := "/proc/sys/net/ipv6/conf/" + entry.Name() + "/accept_ra" + content, err := os.ReadFile(path) + if err != nil { + continue + } + value := strings.TrimSpace(string(content)) + if value != "1" { + continue + } + err = os.WriteFile(path, []byte("2"), 0o644) + if err != nil { + logger.Debug(E.Cause(err, "overrule ", path)) + continue + } + restore = append(restore, sysctlState{name: path, value: value}) + } + return restore +} + func restoreBridgeForwarding(states []sysctlState) { for _, state := range states { _ = os.WriteFile(state.name, []byte(state.value), 0o644) diff --git a/protocol/group/selector.go b/protocol/group/selector.go index 2b1cb04071..e9a2003cc2 100644 --- a/protocol/group/selector.go +++ b/protocol/group/selector.go @@ -162,6 +162,7 @@ func (s *Selector) ListenPacket(ctx context.Context, destination M.Socksaddr) (n func (s *Selector) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { ctx = interrupt.ContextWithIsExternalConnection(ctx) + conn = s.interruptGroup.NewConn(conn, true) selected := s.selected.Load() if outboundHandler, isHandler := selected.(adapter.ConnectionHandler); isHandler { outboundHandler.NewConnection(ctx, conn, metadata, onClose) @@ -172,6 +173,7 @@ func (s *Selector) NewConnection(ctx context.Context, conn net.Conn, metadata ad func (s *Selector) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { ctx = interrupt.ContextWithIsExternalConnection(ctx) + conn = s.interruptGroup.NewSingPacketConn(conn, true) selected := s.selected.Load() if outboundHandler, isHandler := selected.(adapter.PacketConnectionHandler); isHandler { outboundHandler.NewPacketConnection(ctx, conn, metadata, onClose) diff --git a/test/group_test.go b/test/group_test.go new file mode 100644 index 0000000000..2b1ad0bb74 --- /dev/null +++ b/test/group_test.go @@ -0,0 +1,201 @@ +package main + +import ( + "context" + "errors" + "net" + "net/netip" + "testing" + "time" + + C "github.com/sagernet/sing-box/constant" + "github.com/sagernet/sing-box/option" + "github.com/sagernet/sing-box/protocol/group" + "github.com/sagernet/sing/common" + "github.com/sagernet/sing/common/json/badoption" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + "github.com/sagernet/sing/protocol/socks" + + "github.com/stretchr/testify/require" +) + +func startSelectorInstance(t *testing.T, interruptExistConnections bool) *group.Selector { + instance := startInstance(t, option.Options{ + Inbounds: []option.Inbound{ + { + Type: C.TypeSOCKS, + Tag: "socks-in", + Options: &option.SocksInboundOptions{ + ListenOptions: option.ListenOptions{ + Listen: common.Ptr(badoption.Addr(netip.IPv4Unspecified())), + ListenPort: clientPort, + }, + }, + }, + }, + Outbounds: []option.Outbound{ + { + Type: C.TypeSelector, + Tag: "sel", + Options: &option.SelectorOutboundOptions{ + Outbounds: []string{"memA", "memB"}, + Default: "memA", + InterruptExistConnections: interruptExistConnections, + }, + }, + { + Type: C.TypeDirect, + Tag: "memA", + }, + { + Type: C.TypeDirect, + Tag: "memB", + }, + }, + Route: &option.RouteOptions{ + Final: "sel", + }, + }) + outbound, loaded := instance.Outbound().Outbound("sel") + require.True(t, loaded) + selector, isSelector := outbound.(*group.Selector) + require.True(t, isSelector) + return selector +} + +func startTCPEchoServer(t *testing.T, port uint16) { + listener, err := net.ListenTCP("tcp", &net.TCPAddr{IP: localIP.AsSlice(), Port: int(port)}) + require.NoError(t, err) + t.Cleanup(func() { + listener.Close() + }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + go func() { + buffer := make([]byte, 1024) + for { + n, err := conn.Read(buffer) + if err != nil { + conn.Close() + return + } + _, err = conn.Write(buffer[:n]) + if err != nil { + conn.Close() + return + } + } + }() + } + }() +} + +func startUDPEchoServer(t *testing.T, port uint16) chan net.Addr { + packetConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: localIP.AsSlice(), Port: int(port)}) + require.NoError(t, err) + t.Cleanup(func() { + packetConn.Close() + }) + sourceCh := make(chan net.Addr, 16) + go func() { + buffer := make([]byte, 1024) + for { + n, addr, err := packetConn.ReadFrom(buffer) + if err != nil { + close(sourceCh) + return + } + sourceCh <- addr + _, err = packetConn.WriteTo(buffer[:n], addr) + if err != nil { + return + } + } + }() + return sourceCh +} + +func tcpPingPong(t *testing.T, conn net.Conn) { + _, err := conn.Write([]byte("ping")) + require.NoError(t, err) + buffer := make([]byte, 4) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(5*time.Second))) + _, err = conn.Read(buffer) + require.NoError(t, err) + require.Equal(t, []byte("ping"), buffer) + require.NoError(t, conn.SetReadDeadline(time.Time{})) +} + +func TestSelectorInterruptRoutedConnection(t *testing.T) { + selector := startSelectorInstance(t, true) + startTCPEchoServer(t, testPort) + + dialer := socks.NewClient(N.SystemDialer, M.ParseSocksaddrHostPort("127.0.0.1", clientPort), socks.Version5, "", "") + conn, err := dialer.DialContext(context.Background(), "tcp", M.ParseSocksaddrHostPort("127.0.0.1", testPort)) + require.NoError(t, err) + defer conn.Close() + tcpPingPong(t, conn) + + require.True(t, selector.SelectOutbound("memB")) + + require.NoError(t, conn.SetReadDeadline(time.Now().Add(3*time.Second))) + _, err = conn.Read(make([]byte, 4)) + require.Error(t, err, "existing connection routed through selector must be interrupted") + var netErr net.Error + if errors.As(err, &netErr) { + require.False(t, netErr.Timeout(), "existing connection routed through selector must be interrupted") + } +} + +func TestSelectorInterruptRoutedPacketConnection(t *testing.T) { + selector := startSelectorInstance(t, true) + sourceCh := startUDPEchoServer(t, testPort) + + dialer := socks.NewClient(N.SystemDialer, M.ParseSocksaddrHostPort("127.0.0.1", clientPort), socks.Version5, "", "") + packetConn, err := dialer.ListenPacket(context.Background(), M.ParseSocksaddrHostPort("127.0.0.1", testPort)) + require.NoError(t, err) + defer packetConn.Close() + + serverAddr := &net.UDPAddr{IP: localIP.AsSlice(), Port: int(testPort)} + buffer := make([]byte, 1024) + + _, err = packetConn.WriteTo([]byte("ping"), serverAddr) + require.NoError(t, err) + require.NoError(t, packetConn.SetReadDeadline(time.Now().Add(5*time.Second))) + _, _, err = packetConn.ReadFrom(buffer) + require.NoError(t, err) + firstSource := <-sourceCh + + require.True(t, selector.SelectOutbound("memB")) + // The interrupted NAT session cannot signal the SOCKS client; instead verify + // that a subsequent packet no longer flows over the old outbound socket. + _, err = packetConn.WriteTo([]byte("ping"), serverAddr) + require.NoError(t, err) + require.NoError(t, packetConn.SetReadDeadline(time.Now().Add(3*time.Second))) + _, _, err = packetConn.ReadFrom(buffer) + if err == nil { + secondSource := <-sourceCh + require.NotEqual(t, firstSource.String(), secondSource.String(), + "packet connection routed through selector must be interrupted") + } +} + +func TestSelectorKeepRoutedConnection(t *testing.T) { + selector := startSelectorInstance(t, false) + startTCPEchoServer(t, testPort) + + dialer := socks.NewClient(N.SystemDialer, M.ParseSocksaddrHostPort("127.0.0.1", clientPort), socks.Version5, "", "") + conn, err := dialer.DialContext(context.Background(), "tcp", M.ParseSocksaddrHostPort("127.0.0.1", testPort)) + require.NoError(t, err) + defer conn.Close() + tcpPingPong(t, conn) + + require.True(t, selector.SelectOutbound("memB")) + + tcpPingPong(t, conn) +}