cloudflare: enforce socks-proxy ip_rules

protocol/cloudflare/dispatch.go

@@ -272,7 +272,7 @@ func (i *Inbound) handleHTTPService(ctx context.Context, stream io.ReadWriteClos
 			respWriter.WriteResponse(err, nil)
 			return
 		}
-		i.handleSocksProxyStream(ctx, stream, respWriter, request, metadata)
+		i.handleSocksProxyStream(ctx, stream, respWriter, request, metadata, service)
 	default:
 		err := E.New("unsupported service kind for HTTP/WebSocket request")
 		i.logger.ErrorContext(ctx, err)

protocol/cloudflare/ip_rule_policy.go

@@ -0,0 +1,96 @@
+//go:build with_cloudflare_tunnel
+
+package cloudflare
+
+import (
+	"context"
+	"net"
+	"net/netip"
+	"sort"
+
+	E "github.com/sagernet/sing/common/exceptions"
+	M "github.com/sagernet/sing/common/metadata"
+)
+
+type compiledIPRule struct {
+	prefix netip.Prefix
+	ports  []int
+	allow  bool
+}
+
+type ipRulePolicy struct {
+	rules []compiledIPRule
+}
+
+func newIPRulePolicy(rawRules []IPRule) (*ipRulePolicy, error) {
+	policy := &ipRulePolicy{
+		rules: make([]compiledIPRule, 0, len(rawRules)),
+	}
+	for _, rawRule := range rawRules {
+		if rawRule.Prefix == "" {
+			return nil, E.New("ip_rule prefix cannot be blank")
+		}
+		prefix, err := netip.ParsePrefix(rawRule.Prefix)
+		if err != nil {
+			return nil, E.Cause(err, "parse ip_rule prefix")
+		}
+		ports := append([]int(nil), rawRule.Ports...)
+		sort.Ints(ports)
+		for _, port := range ports {
+			if port < 1 || port > 65535 {
+				return nil, E.New("invalid ip_rule port: ", port)
+			}
+		}
+		policy.rules = append(policy.rules, compiledIPRule{
+			prefix: prefix,
+			ports:  ports,
+			allow:  rawRule.Allow,
+		})
+	}
+	return policy, nil
+}
+
+func (p *ipRulePolicy) Allow(ctx context.Context, destination M.Socksaddr) (bool, error) {
+	if p == nil {
+		return false, nil
+	}
+	ipAddr, err := resolvePolicyDestination(ctx, destination)
+	if err != nil {
+		return false, err
+	}
+	port := int(destination.Port)
+	for _, rule := range p.rules {
+		if !rule.prefix.Contains(ipAddr) {
+			continue
+		}
+		if len(rule.ports) == 0 {
+			return rule.allow, nil
+		}
+		portIndex := sort.SearchInts(rule.ports, port)
+		if portIndex < len(rule.ports) && rule.ports[portIndex] == port {
+			return rule.allow, nil
+		}
+	}
+	return false, nil
+}
+
+func resolvePolicyDestination(ctx context.Context, destination M.Socksaddr) (netip.Addr, error) {
+	if destination.IsIP() {
+		return destination.Unwrap().Addr, nil
+	}
+	if !destination.IsFqdn() {
+		return netip.Addr{}, E.New("destination is neither IP nor FQDN")
+	}
+	ipAddrs, err := net.DefaultResolver.LookupIPAddr(ctx, destination.Fqdn)
+	if err != nil {
+		return netip.Addr{}, E.Cause(err, "resolve destination")
+	}
+	if len(ipAddrs) == 0 {
+		return netip.Addr{}, E.New("resolved destination is empty")
+	}
+	resolvedAddr, ok := netip.AddrFromSlice(ipAddrs[0].IP)
+	if !ok {
+		return netip.Addr{}, E.New("resolved destination is invalid")
+	}
+	return resolvedAddr.Unmap(), nil
+}

protocol/cloudflare/runtime_config.go

@@ -50,6 +50,7 @@ type ResolvedService struct {
 	BaseURL       *url.URL
 	UnixPath      string
 	StatusCode    int
+	SocksPolicy   *ipRulePolicy
 	OriginRequest OriginRequestConfig
 }
 
@@ -432,9 +433,14 @@ func parseResolvedService(rawService string, originRequest OriginRequestConfig)
 			OriginRequest: originRequest,
 		}, nil
 	case rawService == "socks-proxy":
+		policy, err := newIPRulePolicy(originRequest.IPRules)
+		if err != nil {
+			return ResolvedService{}, E.Cause(err, "compile socks-proxy ip rules")
+		}
 		return ResolvedService{
 			Kind:          ResolvedServiceSocksProxy,
 			Service:       rawService,
+			SocksPolicy:   policy,
 			OriginRequest: originRequest,
 		}, nil
 	case strings.HasPrefix(rawService, "unix:"):

protocol/cloudflare/special_service.go

@@ -28,6 +28,13 @@ import (
 
 var wsAcceptGUID = []byte("258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
 
+const (
+	socksReplySuccess             = 0
+	socksReplyRuleFailure         = 2
+	socksReplyHostUnreachable     = 4
+	socksReplyCommandNotSupported = 7
+)
+
 func (i *Inbound) handleBastionStream(ctx context.Context, stream io.ReadWriteCloser, respWriter ConnectResponseWriter, request *ConnectRequest, metadata adapter.InboundContext) {
 	destination, err := resolveBastionDestination(request)
 	if err != nil {
@@ -60,7 +67,7 @@ func (i *Inbound) handleRouterBackedStream(ctx context.Context, stream io.ReadWr
 	_ = bufio.CopyConn(ctx, wsConn, targetConn)
 }
 
-func (i *Inbound) handleSocksProxyStream(ctx context.Context, stream io.ReadWriteCloser, respWriter ConnectResponseWriter, request *ConnectRequest, metadata adapter.InboundContext) {
+func (i *Inbound) handleSocksProxyStream(ctx context.Context, stream io.ReadWriteCloser, respWriter ConnectResponseWriter, request *ConnectRequest, metadata adapter.InboundContext, service ResolvedService) {
 	err := respWriter.WriteResponse(nil, encodeResponseHeaders(http.StatusSwitchingProtocols, websocketResponseHeaders(request)))
 	if err != nil {
 		i.logger.ErrorContext(ctx, "write socks-proxy websocket response: ", err)
@@ -69,7 +76,7 @@ func (i *Inbound) handleSocksProxyStream(ctx context.Context, stream io.ReadWrit
 
 	wsConn := v2raywebsocket.NewConn(newStreamConn(stream), nil, ws.StateServerSide)
 	defer wsConn.Close()
-	if err := i.serveSocksProxy(ctx, wsConn); err != nil && !E.IsClosedOrCanceled(err) {
+	if err := i.serveSocksProxy(ctx, wsConn, service.SocksPolicy); err != nil && !E.IsClosedOrCanceled(err) {
 		i.logger.DebugContext(ctx, "socks-proxy stream closed: ", err)
 	}
 }
@@ -132,7 +139,7 @@ func (i *Inbound) dialRouterTCP(ctx context.Context, destination M.Socksaddr) (n
 	}, nil
 }
 
-func (i *Inbound) serveSocksProxy(ctx context.Context, conn net.Conn) error {
+func (i *Inbound) serveSocksProxy(ctx context.Context, conn net.Conn, policy *ipRulePolicy) error {
 	version := make([]byte, 1)
 	if _, err := io.ReadFull(conn, version); err != nil {
 		return err
@@ -161,7 +168,7 @@ func (i *Inbound) serveSocksProxy(ctx context.Context, conn net.Conn) error {
 		return E.New("unsupported SOCKS request version: ", requestHeader[0])
 	}
 	if requestHeader[1] != 1 {
-		_, _ = conn.Write([]byte{5, 7, 0, 1, 0, 0, 0, 0, 0, 0})
+		_ = writeSocksReply(conn, socksReplyCommandNotSupported)
 		return E.New("unsupported SOCKS command: ", requestHeader[1])
 	}
 
@@ -169,19 +176,33 @@ func (i *Inbound) serveSocksProxy(ctx context.Context, conn net.Conn) error {
 	if err != nil {
 		return err
 	}
+	allowed, err := policy.Allow(ctx, destination)
+	if err != nil {
+		_ = writeSocksReply(conn, socksReplyRuleFailure)
+		return err
+	}
+	if !allowed {
+		_ = writeSocksReply(conn, socksReplyRuleFailure)
+		return E.New("connect to ", destination, " denied by ip_rules")
+	}
 	targetConn, cleanup, err := i.dialRouterTCP(ctx, destination)
 	if err != nil {
-		_, _ = conn.Write([]byte{5, 4, 0, 1, 0, 0, 0, 0, 0, 0})
+		_ = writeSocksReply(conn, socksReplyHostUnreachable)
 		return err
 	}
 	defer cleanup()
 
-	if _, err := conn.Write([]byte{5, 0, 0, 1, 0, 0, 0, 0, 0, 0}); err != nil {
+	if err := writeSocksReply(conn, socksReplySuccess); err != nil {
 		return err
 	}
 	return bufio.CopyConn(ctx, conn, targetConn)
 }
 
+func writeSocksReply(conn net.Conn, reply byte) error {
+	_, err := conn.Write([]byte{5, reply, 0, 1, 0, 0, 0, 0, 0, 0})
+	return err
+}
+
 func readSocksDestination(conn net.Conn, addressType byte) (M.Socksaddr, error) {
 	switch addressType {
 	case 1:

protocol/cloudflare/special_service_test.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"strconv"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,6 +18,7 @@ import (
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
 	M "github.com/sagernet/sing/common/metadata"
+	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/ws"
 	"github.com/sagernet/ws/wsutil"
 )
@@ -48,6 +50,10 @@ func (w *fakeConnectResponseWriter) WriteResponse(responseError error, metadata
 }
 
 func newSpecialServiceInbound(t *testing.T) *Inbound {
+	return newSpecialServiceInboundWithRouter(t, &testRouter{})
+}
+
+func newSpecialServiceInboundWithRouter(t *testing.T, router adapter.Router) *Inbound {
 	t.Helper()
 	logFactory, err := log.New(log.Options{Options: option.LogOptions{Level: "debug"}})
 	if err != nil {
@@ -59,19 +65,28 @@ func newSpecialServiceInbound(t *testing.T) *Inbound {
 	}
 	return &Inbound{
 		Adapter:       inbound.NewAdapter(C.TypeCloudflareTunnel, "test"),
-		router:        &testRouter{},
+		router:        router,
 		logger:        logFactory.NewLogger("test"),
 		configManager: configManager,
 	}
 }
 
-func TestHandleBastionStream(t *testing.T) {
+type countingRouter struct {
+	testRouter
+	count atomic.Int32
+}
+
+func (r *countingRouter) RouteConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	r.count.Add(1)
+	r.testRouter.RouteConnectionEx(ctx, conn, metadata, onClose)
+}
+
+func startEchoListener(t *testing.T) net.Listener {
+	t.Helper()
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer listener.Close()
-
 	go func() {
 		for {
 			conn, err := listener.Accept()
@@ -84,6 +99,92 @@ func TestHandleBastionStream(t *testing.T) {
 			}(conn)
 		}
 	}()
+	return listener
+}
+
+func newSocksProxyService(t *testing.T, rules []option.CloudflareTunnelIPRule) ResolvedService {
+	t.Helper()
+	service, err := parseResolvedService("socks-proxy", originRequestFromOption(option.CloudflareTunnelOriginRequestOptions{
+		IPRules: rules,
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return service
+}
+
+func newSocksProxyConnectRequest() *ConnectRequest {
+	return &ConnectRequest{
+		Type: ConnectionTypeWebsocket,
+		Metadata: []Metadata{
+			{Key: metadataHTTPHeader + ":Sec-WebSocket-Key", Val: "dGhlIHNhbXBsZSBub25jZQ=="},
+		},
+	}
+}
+
+func startSocksProxyStream(t *testing.T, inboundInstance *Inbound, service ResolvedService) (net.Conn, <-chan struct{}) {
+	t.Helper()
+	serverSide, clientSide := net.Pipe()
+	respWriter := &fakeConnectResponseWriter{done: make(chan struct{})}
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		inboundInstance.handleSocksProxyStream(context.Background(), serverSide, respWriter, newSocksProxyConnectRequest(), adapter.InboundContext{}, service)
+	}()
+	select {
+	case <-respWriter.done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for socks-proxy connect response")
+	}
+	if respWriter.err != nil {
+		t.Fatal(respWriter.err)
+	}
+	if respWriter.status != http.StatusSwitchingProtocols {
+		t.Fatalf("expected 101 response, got %d", respWriter.status)
+	}
+	return clientSide, done
+}
+
+func writeSocksAuth(t *testing.T, conn net.Conn) {
+	t.Helper()
+	if err := wsutil.WriteClientMessage(conn, ws.OpBinary, []byte{5, 1, 0}); err != nil {
+		t.Fatal(err)
+	}
+	data, _, err := wsutil.ReadServerData(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(data) != string([]byte{5, 0}) {
+		t.Fatalf("unexpected auth response: %v", data)
+	}
+}
+
+func writeSocksConnectIPv4(t *testing.T, conn net.Conn, address string) []byte {
+	t.Helper()
+	host, portText, err := net.SplitHostPort(address)
+	if err != nil {
+		t.Fatal(err)
+	}
+	port, err := strconv.Atoi(portText)
+	if err != nil {
+		t.Fatal(err)
+	}
+	requestBytes := []byte{5, 1, 0, 1}
+	requestBytes = append(requestBytes, net.ParseIP(host).To4()...)
+	requestBytes = append(requestBytes, byte(port>>8), byte(port))
+	if err := wsutil.WriteClientMessage(conn, ws.OpBinary, requestBytes); err != nil {
+		t.Fatal(err)
+	}
+	data, _, err := wsutil.ReadServerData(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return data
+}
+
+func TestHandleBastionStream(t *testing.T) {
+	listener := startEchoListener(t)
+	defer listener.Close()
 
 	serverSide, clientSide := net.Pipe()
 	defer clientSide.Close()
@@ -141,78 +242,22 @@ func TestHandleBastionStream(t *testing.T) {
 }
 
 func TestHandleSocksProxyStream(t *testing.T) {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatal(err)
-	}
+	listener := startEchoListener(t)
 	defer listener.Close()
 
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				return
-			}
-			go func(conn net.Conn) {
-				defer conn.Close()
-				_, _ = io.Copy(conn, conn)
-			}(conn)
-		}
-	}()
+	_, portText, _ := net.SplitHostPort(listener.Addr().String())
+	port, _ := strconv.Atoi(portText)
+	service := newSocksProxyService(t, []option.CloudflareTunnelIPRule{{
+		Prefix: "127.0.0.0/8",
+		Ports:  []int{port},
+		Allow:  true,
+	}})
 
-	serverSide, clientSide := net.Pipe()
+	clientSide, done := startSocksProxyStream(t, newSpecialServiceInbound(t), service)
 	defer clientSide.Close()
 
-	inboundInstance := newSpecialServiceInbound(t)
-	request := &ConnectRequest{
-		Type: ConnectionTypeWebsocket,
-		Metadata: []Metadata{
-			{Key: metadataHTTPHeader + ":Sec-WebSocket-Key", Val: "dGhlIHNhbXBsZSBub25jZQ=="},
-		},
-	}
-	respWriter := &fakeConnectResponseWriter{done: make(chan struct{})}
-
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		inboundInstance.handleSocksProxyStream(context.Background(), serverSide, respWriter, request, adapter.InboundContext{})
-	}()
-
-	select {
-	case <-respWriter.done:
-	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for socks-proxy connect response")
-	}
-	if respWriter.err != nil {
-		t.Fatal(respWriter.err)
-	}
-	if respWriter.status != http.StatusSwitchingProtocols {
-		t.Fatalf("expected 101 response, got %d", respWriter.status)
-	}
-
-	if err := wsutil.WriteClientMessage(clientSide, ws.OpBinary, []byte{5, 1, 0}); err != nil {
-		t.Fatal(err)
-	}
-	data, _, err := wsutil.ReadServerData(clientSide)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if string(data) != string([]byte{5, 0}) {
-		t.Fatalf("unexpected auth response: %v", data)
-	}
-
-	host, portText, _ := net.SplitHostPort(listener.Addr().String())
-	port, _ := strconv.Atoi(portText)
-	requestBytes := []byte{5, 1, 0, 1}
-	requestBytes = append(requestBytes, net.ParseIP(host).To4()...)
-	requestBytes = append(requestBytes, byte(port>>8), byte(port))
-	if err := wsutil.WriteClientMessage(clientSide, ws.OpBinary, requestBytes); err != nil {
-		t.Fatal(err)
-	}
-	data, _, err = wsutil.ReadServerData(clientSide)
-	if err != nil {
-		t.Fatal(err)
-	}
+	writeSocksAuth(t, clientSide)
+	data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
 	if len(data) != 10 || data[1] != 0 {
 		t.Fatalf("unexpected connect response: %v", data)
 	}
@@ -220,7 +265,7 @@ func TestHandleSocksProxyStream(t *testing.T) {
 	if err := wsutil.WriteClientMessage(clientSide, ws.OpBinary, []byte("hello")); err != nil {
 		t.Fatal(err)
 	}
-	data, _, err = wsutil.ReadServerData(clientSide)
+	data, _, err := wsutil.ReadServerData(clientSide)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -235,25 +280,149 @@ func TestHandleSocksProxyStream(t *testing.T) {
 	}
 }
 
-func TestHandleStreamService(t *testing.T) {
-	listener, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatal(err)
-	}
+func TestHandleSocksProxyStreamDenyRule(t *testing.T) {
+	listener := startEchoListener(t)
 	defer listener.Close()
 
-	go func() {
-		for {
-			conn, err := listener.Accept()
-			if err != nil {
-				return
-			}
-			go func(conn net.Conn) {
-				defer conn.Close()
-				_, _ = io.Copy(conn, conn)
-			}(conn)
+	_, portText, _ := net.SplitHostPort(listener.Addr().String())
+	port, _ := strconv.Atoi(portText)
+	service := newSocksProxyService(t, []option.CloudflareTunnelIPRule{{
+		Prefix: "127.0.0.0/8",
+		Ports:  []int{port},
+		Allow:  false,
+	}})
+	router := &countingRouter{}
+	clientSide, done := startSocksProxyStream(t, newSpecialServiceInboundWithRouter(t, router), service)
+	defer clientSide.Close()
+
+	writeSocksAuth(t, clientSide)
+	data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
+	if len(data) != 10 || data[1] != socksReplyRuleFailure {
+		t.Fatalf("unexpected deny response: %v", data)
+	}
+	if router.count.Load() != 0 {
+		t.Fatalf("expected no router dial, got %d", router.count.Load())
+	}
+	_ = clientSide.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("socks-proxy stream did not exit")
+	}
+}
+
+func TestHandleSocksProxyStreamPortMismatchDefaultDeny(t *testing.T) {
+	listener := startEchoListener(t)
+	defer listener.Close()
+
+	_, portText, _ := net.SplitHostPort(listener.Addr().String())
+	port, _ := strconv.Atoi(portText)
+	service := newSocksProxyService(t, []option.CloudflareTunnelIPRule{{
+		Prefix: "127.0.0.0/8",
+		Ports:  []int{port + 1},
+		Allow:  true,
+	}})
+	router := &countingRouter{}
+	clientSide, done := startSocksProxyStream(t, newSpecialServiceInboundWithRouter(t, router), service)
+	defer clientSide.Close()
+
+	writeSocksAuth(t, clientSide)
+	data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
+	if len(data) != 10 || data[1] != socksReplyRuleFailure {
+		t.Fatalf("unexpected port mismatch response: %v", data)
+	}
+	if router.count.Load() != 0 {
+		t.Fatalf("expected no router dial, got %d", router.count.Load())
+	}
+	_ = clientSide.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("socks-proxy stream did not exit")
+	}
+}
+
+func TestHandleSocksProxyStreamEmptyRulesDefaultDeny(t *testing.T) {
+	listener := startEchoListener(t)
+	defer listener.Close()
+
+	router := &countingRouter{}
+	clientSide, done := startSocksProxyStream(t, newSpecialServiceInboundWithRouter(t, router), newSocksProxyService(t, nil))
+	defer clientSide.Close()
+
+	writeSocksAuth(t, clientSide)
+	data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
+	if len(data) != 10 || data[1] != socksReplyRuleFailure {
+		t.Fatalf("unexpected empty-rule response: %v", data)
+	}
+	if router.count.Load() != 0 {
+		t.Fatalf("expected no router dial, got %d", router.count.Load())
+	}
+	_ = clientSide.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("socks-proxy stream did not exit")
+	}
+}
+
+func TestHandleSocksProxyStreamRuleOrderFirstMatchWins(t *testing.T) {
+	listener := startEchoListener(t)
+	defer listener.Close()
+
+	_, portText, _ := net.SplitHostPort(listener.Addr().String())
+	port, _ := strconv.Atoi(portText)
+	allowFirst := newSocksProxyService(t, []option.CloudflareTunnelIPRule{
+		{Prefix: "127.0.0.0/8", Ports: []int{port}, Allow: true},
+		{Prefix: "127.0.0.1/32", Ports: []int{port}, Allow: false},
+	})
+	denyFirst := newSocksProxyService(t, []option.CloudflareTunnelIPRule{
+		{Prefix: "127.0.0.1/32", Ports: []int{port}, Allow: false},
+		{Prefix: "127.0.0.0/8", Ports: []int{port}, Allow: true},
+	})
+
+	t.Run("allow-first", func(t *testing.T) {
+		clientSide, done := startSocksProxyStream(t, newSpecialServiceInbound(t), allowFirst)
+		defer clientSide.Close()
+
+		writeSocksAuth(t, clientSide)
+		data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
+		if len(data) != 10 || data[1] != socksReplySuccess {
+			t.Fatalf("unexpected allow-first response: %v", data)
 		}
-	}()
+		_ = clientSide.Close()
+		select {
+		case <-done:
+		case <-time.After(2 * time.Second):
+			t.Fatal("socks-proxy stream did not exit")
+		}
+	})
+
+	t.Run("deny-first", func(t *testing.T) {
+		router := &countingRouter{}
+		clientSide, done := startSocksProxyStream(t, newSpecialServiceInboundWithRouter(t, router), denyFirst)
+		defer clientSide.Close()
+
+		writeSocksAuth(t, clientSide)
+		data := writeSocksConnectIPv4(t, clientSide, listener.Addr().String())
+		if len(data) != 10 || data[1] != socksReplyRuleFailure {
+			t.Fatalf("unexpected deny-first response: %v", data)
+		}
+		if router.count.Load() != 0 {
+			t.Fatalf("expected no router dial, got %d", router.count.Load())
+		}
+		_ = clientSide.Close()
+		select {
+		case <-done:
+		case <-time.After(2 * time.Second):
+			t.Fatal("socks-proxy stream did not exit")
+		}
+	})
+}
+
+func TestHandleStreamService(t *testing.T) {
+	listener := startEchoListener(t)
+	defer listener.Close()
 
 	serverSide, clientSide := net.Pipe()
 	defer clientSide.Close()