Initial plan

client/proxy/proxy.go

@@ -16,7 +16,6 @@ package proxy
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net"
 	"reflect"
@@ -123,33 +122,6 @@ func (pxy *BaseProxy) Close() {
 	}
 }
 
-// wrapWorkConn applies rate limiting, encryption, and compression
-// to a work connection based on the proxy's transport configuration.
-// The returned recycle function should be called when the stream is no longer in use
-// to return compression resources to the pool. It is safe to not call recycle,
-// in which case resources will be garbage collected normally.
-func (pxy *BaseProxy) wrapWorkConn(conn net.Conn, encKey []byte) (io.ReadWriteCloser, func(), error) {
-	var rwc io.ReadWriteCloser = conn
-	if pxy.limiter != nil {
-		rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error {
-			return conn.Close()
-		})
-	}
-	if pxy.baseCfg.Transport.UseEncryption {
-		var err error
-		rwc, err = libio.WithEncryption(rwc, encKey)
-		if err != nil {
-			conn.Close()
-			return nil, nil, fmt.Errorf("create encryption stream error: %w", err)
-		}
-	}
-	var recycleFn func()
-	if pxy.baseCfg.Transport.UseCompression {
-		rwc, recycleFn = libio.WithCompressionFromPool(rwc)
-	}
-	return rwc, recycleFn, nil
-}
-
 func (pxy *BaseProxy) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) {
 	pxy.inWorkConnCallback = cb
 }
@@ -167,15 +139,31 @@ func (pxy *BaseProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
 func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWorkConn, encKey []byte) {
 	xl := pxy.xl
 	baseCfg := pxy.baseCfg
+	var (
+		remote io.ReadWriteCloser
+		err    error
+	)
+	remote = workConn
+	if pxy.limiter != nil {
+		remote = libio.WrapReadWriteCloser(limit.NewReader(workConn, pxy.limiter), limit.NewWriter(workConn, pxy.limiter), func() error {
+			return workConn.Close()
+		})
+	}
 
 	xl.Tracef("handle tcp work connection, useEncryption: %t, useCompression: %t",
 		baseCfg.Transport.UseEncryption, baseCfg.Transport.UseCompression)
-
+	if baseCfg.Transport.UseEncryption {
-	remote, recycleFn, err := pxy.wrapWorkConn(workConn, encKey)
+		remote, err = libio.WithEncryption(remote, encKey)
 		if err != nil {
-		xl.Errorf("wrap work connection: %v", err)
+			workConn.Close()
+			xl.Errorf("create encryption stream error: %v", err)
 			return
 		}
+	}
+	var compressionResourceRecycleFn func()
+	if baseCfg.Transport.UseCompression {
+		remote, compressionResourceRecycleFn = libio.WithCompressionFromPool(remote)
+	}
 
 	// check if we need to send proxy protocol info
 	var connInfo plugin.ConnectionInfo
@@ -190,6 +178,7 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
 	}
 
 	if baseCfg.Transport.ProxyProtocolVersion != "" && m.SrcAddr != "" && m.SrcPort != 0 {
+		// Use the common proxy protocol builder function
 		header := netpkg.BuildProxyProtocolHeaderStruct(connInfo.SrcAddr, connInfo.DstAddr, baseCfg.Transport.ProxyProtocolVersion)
 		connInfo.ProxyProtocolHeader = header
 	}
@@ -198,18 +187,12 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
 
 	if pxy.proxyPlugin != nil {
 		// if plugin is set, let plugin handle connection first
-		// Don't recycle compression resources here because plugins may
-		// retain the connection after Handle returns.
 		xl.Debugf("handle by plugin: %s", pxy.proxyPlugin.Name())
 		pxy.proxyPlugin.Handle(pxy.ctx, &connInfo)
 		xl.Debugf("handle by plugin finished")
 		return
 	}
 
-	if recycleFn != nil {
-		defer recycleFn()
-	}
-
 	localConn, err := libnet.Dial(
 		net.JoinHostPort(baseCfg.LocalIP, strconv.Itoa(baseCfg.LocalPort)),
 		libnet.WithTimeout(10*time.Second),
@@ -226,7 +209,6 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
 	if connInfo.ProxyProtocolHeader != nil {
 		if _, err := connInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
 			workConn.Close()
-			localConn.Close()
 			xl.Errorf("write proxy protocol header to local conn error: %v", err)
 			return
 		}
@@ -237,4 +219,7 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
 	if len(errs) > 0 {
 		xl.Tracef("join connections errors: %v", errs)
 	}
+	if compressionResourceRecycleFn != nil {
+		compressionResourceRecycleFn()
+	}
 }

client/proxy/sudp.go

@@ -17,6 +17,7 @@
 package proxy
 
 import (
+	"io"
 	"net"
 	"reflect"
 	"strconv"
@@ -24,10 +25,12 @@ import (
 	"time"
 
 	"github.com/fatedier/golib/errors"
+	libio "github.com/fatedier/golib/io"
 
 	v1 "github.com/fatedier/frp/pkg/config/v1"
 	"github.com/fatedier/frp/pkg/msg"
 	"github.com/fatedier/frp/pkg/proto/udp"
+	"github.com/fatedier/frp/pkg/util/limit"
 	netpkg "github.com/fatedier/frp/pkg/util/net"
 )
 
@@ -80,13 +83,27 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	xl := pxy.xl
 	xl.Infof("incoming a new work connection for sudp proxy, %s", conn.RemoteAddr().String())
 
-	remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
+	var rwc io.ReadWriteCloser = conn
+	var err error
+	if pxy.limiter != nil {
+		rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error {
+			return conn.Close()
+		})
+	}
+	if pxy.cfg.Transport.UseEncryption {
+		rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
 		if err != nil {
-		xl.Errorf("wrap work connection: %v", err)
+			conn.Close()
+			xl.Errorf("create encryption stream error: %v", err)
 			return
 		}
+	}
+	if pxy.cfg.Transport.UseCompression {
+		rwc = libio.WithCompression(rwc)
+	}
+	conn = netpkg.WrapReadWriteCloserToConn(rwc, conn)
 
-	workConn := netpkg.WrapReadWriteCloserToConn(remote, conn)
+	workConn := conn
 	readCh := make(chan *msg.UDPPacket, 1024)
 	sendCh := make(chan msg.Message, 1024)
 	isClose := false

client/proxy/udp.go

@@ -17,16 +17,19 @@
 package proxy
 
 import (
+	"io"
 	"net"
 	"reflect"
 	"strconv"
 	"time"
 
 	"github.com/fatedier/golib/errors"
+	libio "github.com/fatedier/golib/io"
 
 	v1 "github.com/fatedier/frp/pkg/config/v1"
 	"github.com/fatedier/frp/pkg/msg"
 	"github.com/fatedier/frp/pkg/proto/udp"
+	"github.com/fatedier/frp/pkg/util/limit"
 	netpkg "github.com/fatedier/frp/pkg/util/net"
 )
 
@@ -91,14 +94,28 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	// close resources related with old workConn
 	pxy.Close()
 
-	remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
+	var rwc io.ReadWriteCloser = conn
+	var err error
+	if pxy.limiter != nil {
+		rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error {
+			return conn.Close()
+		})
+	}
+	if pxy.cfg.Transport.UseEncryption {
+		rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
 		if err != nil {
-		xl.Errorf("wrap work connection: %v", err)
+			conn.Close()
+			xl.Errorf("create encryption stream error: %v", err)
 			return
 		}
+	}
+	if pxy.cfg.Transport.UseCompression {
+		rwc = libio.WithCompression(rwc)
+	}
+	conn = netpkg.WrapReadWriteCloserToConn(rwc, conn)
 
 	pxy.mu.Lock()
-	pxy.workConn = netpkg.WrapReadWriteCloserToConn(remote, conn)
+	pxy.workConn = conn
 	pxy.readCh = make(chan *msg.UDPPacket, 1024)
 	pxy.sendCh = make(chan msg.Message, 1024)
 	pxy.closed = false

client/visitor/stcp.go

@@ -42,10 +42,10 @@ func (sv *STCPVisitor) Run() (err error) {
 		if err != nil {
 			return
 		}
-		go sv.acceptLoop(sv.l, "stcp local", sv.handleConn)
+		go sv.worker()
 	}
 
-	go sv.acceptLoop(sv.internalLn, "stcp internal", sv.handleConn)
+	go sv.internalConnWorker()
 
 	if sv.plugin != nil {
 		sv.plugin.Start()
@@ -57,6 +57,30 @@ func (sv *STCPVisitor) Close() {
 	sv.BaseVisitor.Close()
 }
 
+func (sv *STCPVisitor) worker() {
+	xl := xlog.FromContextSafe(sv.ctx)
+	for {
+		conn, err := sv.l.Accept()
+		if err != nil {
+			xl.Warnf("stcp local listener closed")
+			return
+		}
+		go sv.handleConn(conn)
+	}
+}
+
+func (sv *STCPVisitor) internalConnWorker() {
+	xl := xlog.FromContextSafe(sv.ctx)
+	for {
+		conn, err := sv.internalLn.Accept()
+		if err != nil {
+			xl.Warnf("stcp internal listener closed")
+			return
+		}
+		go sv.handleConn(conn)
+	}
+}
+
 func (sv *STCPVisitor) handleConn(userConn net.Conn) {
 	xl := xlog.FromContextSafe(sv.ctx)
 	var tunnelErr error

client/visitor/sudp.go

@@ -217,7 +217,6 @@ func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
 	}
 	err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
 	if err != nil {
-		visitorConn.Close()
 		return nil, fmt.Errorf("frpc send newVisitorConnMsg to frps error: %v", err)
 	}
 
@@ -225,13 +224,11 @@ func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
 	_ = visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
 	err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
 	if err != nil {
-		visitorConn.Close()
 		return nil, fmt.Errorf("frpc read newVisitorConnRespMsg error: %v", err)
 	}
 	_ = visitorConn.SetReadDeadline(time.Time{})
 
 	if newVisitorConnRespMsg.Error != "" {
-		visitorConn.Close()
 		return nil, fmt.Errorf("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
 	}
 
@@ -241,7 +238,6 @@ func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
 		remote, err = libio.WithEncryption(remote, []byte(sv.cfg.SecretKey))
 		if err != nil {
 			xl.Errorf("create encryption stream error: %v", err)
-			visitorConn.Close()
 			return nil, err
 		}
 	}

client/visitor/visitor.go

@@ -119,18 +119,6 @@ func (v *BaseVisitor) AcceptConn(conn net.Conn) error {
 	return v.internalLn.PutConn(conn)
 }
 
-func (v *BaseVisitor) acceptLoop(l net.Listener, name string, handleConn func(net.Conn)) {
-	xl := xlog.FromContextSafe(v.ctx)
-	for {
-		conn, err := l.Accept()
-		if err != nil {
-			xl.Warnf("%s listener closed", name)
-			return
-		}
-		go handleConn(conn)
-	}
-}
-
 func (v *BaseVisitor) Close() {
 	if v.l != nil {
 		v.l.Close()

client/visitor/xtcp.go

@@ -65,10 +65,10 @@ func (sv *XTCPVisitor) Run() (err error) {
 		if err != nil {
 			return
 		}
-		go sv.acceptLoop(sv.l, "xtcp local", sv.handleConn)
+		go sv.worker()
 	}
 
-	go sv.acceptLoop(sv.internalLn, "xtcp internal", sv.handleConn)
+	go sv.internalConnWorker()
 	go sv.processTunnelStartEvents()
 	if sv.cfg.KeepTunnelOpen {
 		sv.retryLimiter = rate.NewLimiter(rate.Every(time.Hour/time.Duration(sv.cfg.MaxRetriesAnHour)), sv.cfg.MaxRetriesAnHour)
@@ -93,6 +93,30 @@ func (sv *XTCPVisitor) Close() {
 	}
 }
 
+func (sv *XTCPVisitor) worker() {
+	xl := xlog.FromContextSafe(sv.ctx)
+	for {
+		conn, err := sv.l.Accept()
+		if err != nil {
+			xl.Warnf("xtcp local listener closed")
+			return
+		}
+		go sv.handleConn(conn)
+	}
+}
+
+func (sv *XTCPVisitor) internalConnWorker() {
+	xl := xlog.FromContextSafe(sv.ctx)
+	for {
+		conn, err := sv.internalLn.Accept()
+		if err != nil {
+			xl.Warnf("xtcp internal listener closed")
+			return
+		}
+		go sv.handleConn(conn)
+	}
+}
+
 func (sv *XTCPVisitor) processTunnelStartEvents() {
 	for {
 		select {
@@ -187,7 +211,6 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
 		muxConnRWCloser, err = libio.WithEncryption(muxConnRWCloser, []byte(sv.cfg.SecretKey))
 		if err != nil {
 			xl.Errorf("create encryption stream error: %v", err)
-			tunnelConn.Close()
 			tunnelErr = err
 			return
 		}
@@ -350,7 +373,6 @@ func (ks *KCPTunnelSession) Init(listenConn *net.UDPConn, raddr *net.UDPAddr) er
 	}
 	remote, err := netpkg.NewKCPConnFromUDP(lConn, true, raddr.String())
 	if err != nil {
-		lConn.Close()
 		return fmt.Errorf("create kcp connection from udp connection error: %v", err)
 	}
 

pkg/auth/oidc.go

@@ -23,7 +23,6 @@ import (
 	"net/url"
 	"os"
 	"slices"
-	"sync"
 
 	"github.com/coreos/go-oidc/v3/oidc"
 	"golang.org/x/oauth2"
@@ -206,8 +205,7 @@ type OidcAuthConsumer struct {
 	additionalAuthScopes []v1.AuthScope
 
 	verifier          TokenVerifier
-	mu                sync.RWMutex
+	subjectsFromLogin []string
-	subjectsFromLogin map[string]struct{}
 }
 
 func NewTokenVerifier(cfg v1.AuthOIDCServerConfig) TokenVerifier {
@@ -228,7 +226,7 @@ func NewOidcAuthVerifier(additionalAuthScopes []v1.AuthScope, verifier TokenVeri
 	return &OidcAuthConsumer{
 		additionalAuthScopes: additionalAuthScopes,
 		verifier:             verifier,
-		subjectsFromLogin:    make(map[string]struct{}),
+		subjectsFromLogin:    []string{},
 	}
 }
 
@@ -237,9 +235,9 @@ func (auth *OidcAuthConsumer) VerifyLogin(loginMsg *msg.Login) (err error) {
 	if err != nil {
 		return fmt.Errorf("invalid OIDC token in login: %v", err)
 	}
-	auth.mu.Lock()
+	if !slices.Contains(auth.subjectsFromLogin, token.Subject) {
-	auth.subjectsFromLogin[token.Subject] = struct{}{}
+		auth.subjectsFromLogin = append(auth.subjectsFromLogin, token.Subject)
-	auth.mu.Unlock()
+	}
 	return nil
 }
 
@@ -248,13 +246,11 @@ func (auth *OidcAuthConsumer) verifyPostLoginToken(privilegeKey string) (err err
 	if err != nil {
 		return fmt.Errorf("invalid OIDC token in ping: %v", err)
 	}
-	auth.mu.RLock()
+	if !slices.Contains(auth.subjectsFromLogin, token.Subject) {
-	_, ok := auth.subjectsFromLogin[token.Subject]
-	auth.mu.RUnlock()
-	if !ok {
 		return fmt.Errorf("received different OIDC subject in login and ping. "+
+			"original subjects: %s, "+
 			"new subject: %s",
-			token.Subject)
+			auth.subjectsFromLogin, token.Subject)
 	}
 	return nil
 }

pkg/config/legacy/conversion.go

@@ -171,15 +171,16 @@ func Convert_ServerCommonConf_To_v1(conf *ServerCommonConf) *v1.ServerConfig {
 func transformHeadersFromPluginParams(params map[string]string) v1.HeaderOperations {
 	out := v1.HeaderOperations{}
 	for k, v := range params {
-		k, ok := strings.CutPrefix(k, "plugin_header_")
+		if !strings.HasPrefix(k, "plugin_header_") {
-		if !ok || k == "" {
 			continue
 		}
+		if k = strings.TrimPrefix(k, "plugin_header_"); k != "" {
 			if out.Set == nil {
 				out.Set = make(map[string]string)
 			}
 			out.Set[k] = v
 		}
+	}
 	return out
 }
 

pkg/config/legacy/utils.go

@@ -22,8 +22,8 @@ func GetMapWithoutPrefix(set map[string]string, prefix string) map[string]string
 	m := make(map[string]string)
 
 	for key, value := range set {
-		if trimmed, ok := strings.CutPrefix(key, prefix); ok {
+		if strings.HasPrefix(key, prefix) {
-			m[trimmed] = value
+			m[strings.TrimPrefix(key, prefix)] = value
 		}
 	}
 

pkg/config/source/aggregator.go

@@ -15,11 +15,9 @@
 package source
 
 import (
-	"cmp"
 	"errors"
 	"fmt"
-	"maps"
+	"sort"
-	"slices"
 	"sync"
 
 	v1 "github.com/fatedier/frp/pkg/config/v1"
@@ -99,11 +97,21 @@ func (a *Aggregator) mapsToSortedSlices(
 	proxyMap map[string]v1.ProxyConfigurer,
 	visitorMap map[string]v1.VisitorConfigurer,
 ) ([]v1.ProxyConfigurer, []v1.VisitorConfigurer) {
-	proxies := slices.SortedFunc(maps.Values(proxyMap), func(x, y v1.ProxyConfigurer) int {
+	proxies := make([]v1.ProxyConfigurer, 0, len(proxyMap))
-		return cmp.Compare(x.GetBaseConfig().Name, y.GetBaseConfig().Name)
+	for _, p := range proxyMap {
+		proxies = append(proxies, p)
+	}
+	sort.Slice(proxies, func(i, j int) bool {
+		return proxies[i].GetBaseConfig().Name < proxies[j].GetBaseConfig().Name
 	})
-	visitors := slices.SortedFunc(maps.Values(visitorMap), func(x, y v1.VisitorConfigurer) int {
+
-		return cmp.Compare(x.GetBaseConfig().Name, y.GetBaseConfig().Name)
+	visitors := make([]v1.VisitorConfigurer, 0, len(visitorMap))
+	for _, v := range visitorMap {
+		visitors = append(visitors, v)
+	}
+	sort.Slice(visitors, func(i, j int) bool {
+		return visitors[i].GetBaseConfig().Name < visitors[j].GetBaseConfig().Name
 	})
+
 	return proxies, visitors
 }

pkg/config/source/aggregator_test.go

@@ -196,27 +196,6 @@ func TestAggregator_VisitorMerge(t *testing.T) {
 	require.Len(visitors, 2)
 }
 
-func TestAggregator_Load_ReturnsSortedByName(t *testing.T) {
-	require := require.New(t)
-
-	agg := newTestAggregator(t, nil)
-	err := agg.ConfigSource().ReplaceAll(
-		[]v1.ProxyConfigurer{mockProxy("charlie"), mockProxy("alice"), mockProxy("bob")},
-		[]v1.VisitorConfigurer{mockVisitor("zulu"), mockVisitor("alpha")},
-	)
-	require.NoError(err)
-
-	proxies, visitors, err := agg.Load()
-	require.NoError(err)
-	require.Len(proxies, 3)
-	require.Equal("alice", proxies[0].GetBaseConfig().Name)
-	require.Equal("bob", proxies[1].GetBaseConfig().Name)
-	require.Equal("charlie", proxies[2].GetBaseConfig().Name)
-	require.Len(visitors, 2)
-	require.Equal("alpha", visitors[0].GetBaseConfig().Name)
-	require.Equal("zulu", visitors[1].GetBaseConfig().Name)
-}
-
 func TestAggregator_Load_ReturnsDefensiveCopies(t *testing.T) {
 	require := require.New(t)
 

pkg/config/types/types.go

@@ -70,18 +70,24 @@ func (q *BandwidthQuantity) UnmarshalString(s string) error {
 		f    float64
 		err  error
 	)
-	if fstr, ok := strings.CutSuffix(s, "MB"); ok {
+	switch {
+	case strings.HasSuffix(s, "MB"):
 		base = MB
+		fstr := strings.TrimSuffix(s, "MB")
 		f, err = strconv.ParseFloat(fstr, 64)
-	} else if fstr, ok := strings.CutSuffix(s, "KB"); ok {
-		base = KB
-		f, err = strconv.ParseFloat(fstr, 64)
-	} else {
-		return errors.New("unit not support")
-	}
 		if err != nil {
 			return err
 		}
+	case strings.HasSuffix(s, "KB"):
+		base = KB
+		fstr := strings.TrimSuffix(s, "KB")
+		f, err = strconv.ParseFloat(fstr, 64)
+		if err != nil {
+			return err
+		}
+	default:
+		return errors.New("unit not support")
+	}
 
 	q.s = s
 	q.i = int64(f * float64(base))

pkg/config/types/types_test.go

@@ -39,31 +39,6 @@ func TestBandwidthQuantity(t *testing.T) {
 	require.Equal(`{"b":"1KB","int":5}`, string(buf))
 }
 
-func TestBandwidthQuantity_MB(t *testing.T) {
-	require := require.New(t)
-
-	var w Wrap
-	err := json.Unmarshal([]byte(`{"b":"2MB","int":1}`), &w)
-	require.NoError(err)
-	require.EqualValues(2*MB, w.B.Bytes())
-
-	buf, err := json.Marshal(&w)
-	require.NoError(err)
-	require.Equal(`{"b":"2MB","int":1}`, string(buf))
-}
-
-func TestBandwidthQuantity_InvalidUnit(t *testing.T) {
-	var w Wrap
-	err := json.Unmarshal([]byte(`{"b":"1GB","int":1}`), &w)
-	require.Error(t, err)
-}
-
-func TestBandwidthQuantity_InvalidNumber(t *testing.T) {
-	var w Wrap
-	err := json.Unmarshal([]byte(`{"b":"abcKB","int":1}`), &w)
-	require.Error(t, err)
-}
-
 func TestPortsRangeSlice2String(t *testing.T) {
 	require := require.New(t)
 

pkg/metrics/mem/server.go

@@ -203,7 +203,16 @@ func (m *serverMetrics) GetServer() *ServerStats {
 	return s
 }
 
-func toProxyStats(name string, proxyStats *ProxyStatistics) *ProxyStats {
+func (m *serverMetrics) GetProxiesByType(proxyType string) []*ProxyStats {
+	res := make([]*ProxyStats, 0)
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	for name, proxyStats := range m.info.ProxyStatistics {
+		if proxyStats.ProxyType != proxyType {
+			continue
+		}
+
 		ps := &ProxyStats{
 			Name:            name,
 			Type:            proxyStats.ProxyType,
@@ -219,19 +228,7 @@ func toProxyStats(name string, proxyStats *ProxyStatistics) *ProxyStats {
 		if !proxyStats.LastCloseTime.IsZero() {
 			ps.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
 		}
-	return ps
+		res = append(res, ps)
-}
-
-func (m *serverMetrics) GetProxiesByType(proxyType string) []*ProxyStats {
-	res := make([]*ProxyStats, 0)
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	for name, proxyStats := range m.info.ProxyStatistics {
-		if proxyStats.ProxyType != proxyType {
-			continue
-		}
-		res = append(res, toProxyStats(name, proxyStats))
 	}
 	return res
 }
@@ -244,10 +241,26 @@ func (m *serverMetrics) GetProxiesByTypeAndName(proxyType string, proxyName stri
 		if proxyStats.ProxyType != proxyType {
 			continue
 		}
+
 		if name != proxyName {
 			continue
 		}
-		res = toProxyStats(name, proxyStats)
+
+		res = &ProxyStats{
+			Name:            name,
+			Type:            proxyStats.ProxyType,
+			User:            proxyStats.User,
+			ClientID:        proxyStats.ClientID,
+			TodayTrafficIn:  proxyStats.TrafficIn.TodayCount(),
+			TodayTrafficOut: proxyStats.TrafficOut.TodayCount(),
+			CurConns:        int64(proxyStats.CurConns.Count()),
+		}
+		if !proxyStats.LastStartTime.IsZero() {
+			res.LastStartTime = proxyStats.LastStartTime.Format("01-02 15:04:05")
+		}
+		if !proxyStats.LastCloseTime.IsZero() {
+			res.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
+		}
 		break
 	}
 	return
@@ -259,7 +272,21 @@ func (m *serverMetrics) GetProxyByName(proxyName string) (res *ProxyStats) {
 
 	proxyStats, ok := m.info.ProxyStatistics[proxyName]
 	if ok {
-		res = toProxyStats(proxyName, proxyStats)
+		res = &ProxyStats{
+			Name:            proxyName,
+			Type:            proxyStats.ProxyType,
+			User:            proxyStats.User,
+			ClientID:        proxyStats.ClientID,
+			TodayTrafficIn:  proxyStats.TrafficIn.TodayCount(),
+			TodayTrafficOut: proxyStats.TrafficOut.TodayCount(),
+			CurConns:        int64(proxyStats.CurConns.Count()),
+		}
+		if !proxyStats.LastStartTime.IsZero() {
+			res.LastStartTime = proxyStats.LastStartTime.Format("01-02 15:04:05")
+		}
+		if !proxyStats.LastCloseTime.IsZero() {
+			res.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
+		}
 	}
 	return
 }

pkg/naming/names.go

@@ -16,8 +16,9 @@ func StripUserPrefix(user, name string) string {
 	if user == "" {
 		return name
 	}
-	if trimmed, ok := strings.CutPrefix(name, user+"."); ok {
+	prefix := user + "."
-		return trimmed
+	if strings.HasPrefix(name, prefix) {
+		return strings.TrimPrefix(name, prefix)
 	}
 	return name
 }

pkg/nathole/classify.go

@@ -70,8 +70,12 @@ func ClassifyNATFeature(addresses []string, localIPs []string) (*NatFeature, err
 			continue
 		}
 
-		portMax = max(portMax, portNum)
+		if portNum > portMax {
-		portMin = min(portMin, portNum)
+			portMax = portNum
+		}
+		if portNum < portMin {
+			portMin = portNum
+		}
 		if baseIP != ip {
 			ipChanged = true
 		}

pkg/nathole/controller.go

@@ -152,9 +152,7 @@ func (c *Controller) GenSid() string {
 
 func (c *Controller) HandleVisitor(m *msg.NatHoleVisitor, transporter transport.MessageTransporter, visitorUser string) {
 	if m.PreCheck {
-		c.mu.RLock()
 		cfg, ok := c.clientCfgs[m.ProxyName]
-		c.mu.RUnlock()
 		if !ok {
 			_ = transporter.Send(c.GenNatHoleResponse(m.TransactionID, nil, fmt.Sprintf("xtcp server for [%s] doesn't exist", m.ProxyName)))
 			return

pkg/nathole/nathole.go

@@ -298,13 +298,11 @@ func waitDetectMessage(
 		n, raddr, err := conn.ReadFromUDP(buf)
 		_ = conn.SetReadDeadline(time.Time{})
 		if err != nil {
-			pool.PutBuf(buf)
 			return nil, err
 		}
 		xl.Debugf("get udp message local %s, from %s", conn.LocalAddr(), raddr)
 		var m msg.NatHoleSid
 		if err := DecodeMessageInto(buf[:n], key, &m); err != nil {
-			pool.PutBuf(buf)
 			xl.Warnf("decode sid message error: %v", err)
 			continue
 		}

pkg/plugin/client/http2http.go

@@ -21,7 +21,6 @@ import (
 	stdlog "log"
 	"net/http"
 	"net/http/httputil"
-	"time"
 
 	"github.com/fatedier/golib/pool"
 
@@ -69,7 +68,7 @@ func NewHTTP2HTTPPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugin
 
 	p.s = &http.Server{
 		Handler:           rp,
-		ReadHeaderTimeout: 60 * time.Second,
+		ReadHeaderTimeout: 0,
 	}
 
 	go func() {

pkg/plugin/client/http2https.go

@@ -22,7 +22,6 @@ import (
 	stdlog "log"
 	"net/http"
 	"net/http/httputil"
-	"time"
 
 	"github.com/fatedier/golib/pool"
 
@@ -78,7 +77,7 @@ func NewHTTP2HTTPSPlugin(_ PluginContext, options v1.ClientPluginOptions) (Plugi
 
 	p.s = &http.Server{
 		Handler:           rp,
-		ReadHeaderTimeout: 60 * time.Second,
+		ReadHeaderTimeout: 0,
 	}
 
 	go func() {

pkg/plugin/client/tls2raw.go

@@ -62,13 +62,11 @@ func (p *TLS2RawPlugin) Handle(ctx context.Context, connInfo *ConnectionInfo) {
 
 	if err := tlsConn.Handshake(); err != nil {
 		xl.Warnf("tls handshake error: %v", err)
-		tlsConn.Close()
 		return
 	}
 	rawConn, err := net.Dial("tcp", p.opts.LocalAddr)
 	if err != nil {
 		xl.Warnf("dial to local addr error: %v", err)
-		tlsConn.Close()
 		return
 	}
 

pkg/plugin/client/unix_domain_socket.go

@@ -54,13 +54,10 @@ func (uds *UnixDomainSocketPlugin) Handle(ctx context.Context, connInfo *Connect
 	localConn, err := net.DialUnix("unix", nil, uds.UnixAddr)
 	if err != nil {
 		xl.Warnf("dial to uds %s error: %v", uds.UnixAddr, err)
-		connInfo.Conn.Close()
 		return
 	}
 	if connInfo.ProxyProtocolHeader != nil {
 		if _, err := connInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
-			localConn.Close()
-			connInfo.Conn.Close()
 			return
 		}
 	}

pkg/plugin/server/http.go

@@ -24,7 +24,6 @@ import (
 	"net/http"
 	"net/url"
 	"reflect"
-	"slices"
 	"strings"
 
 	v1 "github.com/fatedier/frp/pkg/config/v1"
@@ -65,7 +64,12 @@ func (p *httpPlugin) Name() string {
 }
 
 func (p *httpPlugin) IsSupport(op string) bool {
-	return slices.Contains(p.options.Ops, op)
+	for _, v := range p.options.Ops {
+		if v == op {
+			return true
+		}
+	}
+	return false
 }
 
 func (p *httpPlugin) Handle(ctx context.Context, op string, content any) (*Response, any, error) {

pkg/proto/udp/udp.go

@@ -85,7 +85,6 @@ func Forwarder(dstAddr *net.UDPAddr, readCh <-chan *msg.UDPPacket, sendCh chan<-
 		}()
 
 		buf := pool.GetBuf(bufSize)
-		defer pool.PutBuf(buf)
 		for {
 			_ = udpConn.SetReadDeadline(time.Now().Add(30 * time.Second))
 			n, _, err := udpConn.ReadFromUDP(buf)

pkg/transport/tls.go

@@ -20,7 +20,6 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/pem"
-	"fmt"
 	"math/big"
 	"os"
 	"time"
@@ -86,9 +85,7 @@ func newCertPool(caPath string) (*x509.CertPool, error) {
 		return nil, err
 	}
 
-	if !pool.AppendCertsFromPEM(caCrt) {
+	pool.AppendCertsFromPEM(caCrt)
-		return nil, fmt.Errorf("failed to parse CA certificate from file %q: no valid PEM certificates found", caPath)
-	}
 
 	return pool, nil
 }

pkg/util/net/udp.go

@@ -168,15 +168,11 @@ func ListenUDP(bindAddr string, bindPort int) (l *UDPListener, err error) {
 		return l, err
 	}
 	readConn, err := net.ListenUDP("udp", udpAddr)
-	if err != nil {
-		return l, err
-	}
 
 	l = &UDPListener{
 		addr:      udpAddr,
 		acceptCh:  make(chan net.Conn),
 		writeCh:   make(chan *UDPPacket, 1000),
-		readConn:  readConn,
 		fakeConns: make(map[string]*FakeUDPConn),
 	}
 

pkg/util/net/websocket.go

@@ -26,7 +26,6 @@ type WebsocketListener struct {
 // ln: tcp listener for websocket connections
 func NewWebsocketListener(ln net.Listener) (wl *WebsocketListener) {
 	wl = &WebsocketListener{
-		ln:       ln,
 		acceptCh: make(chan net.Conn),
 	}
 

pkg/util/vhost/http.go

@@ -266,13 +266,31 @@ func (rp *HTTPReverseProxy) connectHandler(rw http.ResponseWriter, req *http.Req
 	go libio.Join(remote, client)
 }
 
+func parseBasicAuth(auth string) (username, password string, ok bool) {
+	const prefix = "Basic "
+	// Case insensitive prefix match. See Issue 22736.
+	if len(auth) < len(prefix) || !strings.EqualFold(auth[:len(prefix)], prefix) {
+		return
+	}
+	c, err := base64.StdEncoding.DecodeString(auth[len(prefix):])
+	if err != nil {
+		return
+	}
+	cs := string(c)
+	s := strings.IndexByte(cs, ':')
+	if s < 0 {
+		return
+	}
+	return cs[:s], cs[s+1:], true
+}
+
 func (rp *HTTPReverseProxy) injectRequestInfoToCtx(req *http.Request) *http.Request {
 	user := ""
 	// If url host isn't empty, it's a proxy request. Get http user from Proxy-Authorization header.
 	if req.URL.Host != "" {
 		proxyAuth := req.Header.Get("Proxy-Authorization")
 		if proxyAuth != "" {
-			user, _, _ = httppkg.ParseBasicAuth(proxyAuth)
+			user, _, _ = parseBasicAuth(proxyAuth)
 		}
 	}
 	if user == "" {

pkg/util/xlog/xlog.go

@@ -63,12 +63,11 @@ func (l *Logger) AddPrefix(prefix LogPrefix) *Logger {
 	if prefix.Priority <= 0 {
 		prefix.Priority = 10
 	}
-	for i, p := range l.prefixes {
+	for _, p := range l.prefixes {
 		if p.Name == prefix.Name {
 			found = true
-			l.prefixes[i].Value = prefix.Value
+			p.Value = prefix.Value
-			l.prefixes[i].Priority = prefix.Priority
+			p.Priority = prefix.Priority
-			break
 		}
 	}
 	if !found {

server/control.go

@@ -95,33 +95,20 @@ func (cm *ControlManager) Close() error {
 	return nil
 }
 
-// SessionContext encapsulates the input parameters for creating a new Control.
-type SessionContext struct {
-	// all resource managers and controllers
-	RC *controller.ResourceController
-	// proxy manager
-	PxyManager *proxy.Manager
-	// plugin manager
-	PluginManager *plugin.Manager
-	// verifies authentication based on selected method
-	AuthVerifier auth.Verifier
-	// key used for connection encryption
-	EncryptionKey []byte
-	// control connection
-	Conn net.Conn
-	// indicates whether the connection is encrypted
-	ConnEncrypted bool
-	// login message
-	LoginMsg *msg.Login
-	// server configuration
-	ServerCfg *v1.ServerConfig
-	// client registry
-	ClientRegistry *registry.ClientRegistry
-}
-
 type Control struct {
-	// session context
+	// all resource managers and controllers
-	sessionCtx *SessionContext
+	rc *controller.ResourceController
+
+	// proxy manager
+	pxyManager *proxy.Manager
+
+	// plugin manager
+	pluginManager *plugin.Manager
+
+	// verifies authentication based on selected method
+	authVerifier auth.Verifier
+	// key used for connection encryption
+	encryptionKey []byte
 
 	// other components can use this to communicate with client
 	msgTransporter transport.MessageTransporter
@@ -130,6 +117,12 @@ type Control struct {
 	// It provides a channel for sending messages, and you can register handlers to process messages based on their respective types.
 	msgDispatcher *msg.Dispatcher
 
+	// login message
+	loginMsg *msg.Login
+
+	// control connection
+	conn net.Conn
+
 	// work connections
 	workConnCh chan net.Conn
 
@@ -152,37 +145,61 @@ type Control struct {
 
 	mu sync.RWMutex
 
+	// Server configuration information
+	serverCfg *v1.ServerConfig
+
+	clientRegistry *registry.ClientRegistry
+
 	xl     *xlog.Logger
 	ctx    context.Context
 	doneCh chan struct{}
 }
 
-func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, error) {
+// TODO(fatedier): Referencing the implementation of frpc, encapsulate the input parameters as SessionContext.
-	poolCount := sessionCtx.LoginMsg.PoolCount
+func NewControl(
-	if poolCount > int(sessionCtx.ServerCfg.Transport.MaxPoolCount) {
+	ctx context.Context,
-		poolCount = int(sessionCtx.ServerCfg.Transport.MaxPoolCount)
+	rc *controller.ResourceController,
+	pxyManager *proxy.Manager,
+	pluginManager *plugin.Manager,
+	authVerifier auth.Verifier,
+	encryptionKey []byte,
+	ctlConn net.Conn,
+	ctlConnEncrypted bool,
+	loginMsg *msg.Login,
+	serverCfg *v1.ServerConfig,
+) (*Control, error) {
+	poolCount := loginMsg.PoolCount
+	if poolCount > int(serverCfg.Transport.MaxPoolCount) {
+		poolCount = int(serverCfg.Transport.MaxPoolCount)
 	}
 	ctl := &Control{
-		sessionCtx:   sessionCtx,
+		rc:            rc,
+		pxyManager:    pxyManager,
+		pluginManager: pluginManager,
+		authVerifier:  authVerifier,
+		encryptionKey: encryptionKey,
+		conn:          ctlConn,
+		loginMsg:      loginMsg,
 		workConnCh:    make(chan net.Conn, poolCount+10),
 		proxies:       make(map[string]proxy.Proxy),
 		poolCount:     poolCount,
 		portsUsedNum:  0,
-		runID:        sessionCtx.LoginMsg.RunID,
+		runID:         loginMsg.RunID,
+		serverCfg:     serverCfg,
 		xl:            xlog.FromContextSafe(ctx),
 		ctx:           ctx,
 		doneCh:        make(chan struct{}),
 	}
 	ctl.lastPing.Store(time.Now())
 
-	if sessionCtx.ConnEncrypted {
+	if ctlConnEncrypted {
-		cryptoRW, err := netpkg.NewCryptoReadWriter(sessionCtx.Conn, sessionCtx.EncryptionKey)
+		cryptoRW, err := netpkg.NewCryptoReadWriter(ctl.conn, ctl.encryptionKey)
 		if err != nil {
 			return nil, err
 		}
 		ctl.msgDispatcher = msg.NewDispatcher(cryptoRW)
 	} else {
-		ctl.msgDispatcher = msg.NewDispatcher(sessionCtx.Conn)
+		ctl.msgDispatcher = msg.NewDispatcher(ctl.conn)
 	}
 	ctl.registerMsgHandlers()
 	ctl.msgTransporter = transport.NewMessageTransporter(ctl.msgDispatcher)
@@ -196,7 +213,7 @@ func (ctl *Control) Start() {
 		RunID:   ctl.runID,
 		Error:   "",
 	}
-	_ = msg.WriteMsg(ctl.sessionCtx.Conn, loginRespMsg)
+	_ = msg.WriteMsg(ctl.conn, loginRespMsg)
 
 	go func() {
 		for i := 0; i < ctl.poolCount; i++ {
@@ -208,7 +225,7 @@ func (ctl *Control) Start() {
 }
 
 func (ctl *Control) Close() error {
-	ctl.sessionCtx.Conn.Close()
+	ctl.conn.Close()
 	return nil
 }
 
@@ -216,7 +233,7 @@ func (ctl *Control) Replaced(newCtl *Control) {
 	xl := ctl.xl
 	xl.Infof("replaced by client [%s]", newCtl.runID)
 	ctl.runID = ""
-	ctl.sessionCtx.Conn.Close()
+	ctl.conn.Close()
 }
 
 func (ctl *Control) RegisterWorkConn(conn net.Conn) error {
@@ -274,7 +291,7 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
 				return
 			}
 
-		case <-time.After(time.Duration(ctl.sessionCtx.ServerCfg.UserConnTimeout) * time.Second):
+		case <-time.After(time.Duration(ctl.serverCfg.UserConnTimeout) * time.Second):
 			err = fmt.Errorf("timeout trying to get work connection")
 			xl.Warnf("%v", err)
 			return
@@ -287,15 +304,15 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
 }
 
 func (ctl *Control) heartbeatWorker() {
-	if ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout <= 0 {
+	if ctl.serverCfg.Transport.HeartbeatTimeout <= 0 {
 		return
 	}
 
 	xl := ctl.xl
 	go wait.Until(func() {
-		if time.Since(ctl.lastPing.Load().(time.Time)) > time.Duration(ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout)*time.Second {
+		if time.Since(ctl.lastPing.Load().(time.Time)) > time.Duration(ctl.serverCfg.Transport.HeartbeatTimeout)*time.Second {
 			xl.Warnf("heartbeat timeout")
-			ctl.sessionCtx.Conn.Close()
+			ctl.conn.Close()
 			return
 		}
 	}, time.Second, ctl.doneCh)
@@ -313,7 +330,7 @@ func (ctl *Control) worker() {
 	go ctl.msgDispatcher.Run()
 
 	<-ctl.msgDispatcher.Done()
-	ctl.sessionCtx.Conn.Close()
+	ctl.conn.Close()
 
 	ctl.mu.Lock()
 	defer ctl.mu.Unlock()
@@ -325,26 +342,26 @@ func (ctl *Control) worker() {
 
 	for _, pxy := range ctl.proxies {
 		pxy.Close()
-		ctl.sessionCtx.PxyManager.Del(pxy.GetName())
+		ctl.pxyManager.Del(pxy.GetName())
 		metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
 
 		notifyContent := &plugin.CloseProxyContent{
 			User: plugin.UserInfo{
-				User:  ctl.sessionCtx.LoginMsg.User,
+				User:  ctl.loginMsg.User,
-				Metas: ctl.sessionCtx.LoginMsg.Metas,
+				Metas: ctl.loginMsg.Metas,
-				RunID: ctl.sessionCtx.LoginMsg.RunID,
+				RunID: ctl.loginMsg.RunID,
 			},
 			CloseProxy: msg.CloseProxy{
 				ProxyName: pxy.GetName(),
 			},
 		}
 		go func() {
-			_ = ctl.sessionCtx.PluginManager.CloseProxy(notifyContent)
+			_ = ctl.pluginManager.CloseProxy(notifyContent)
 		}()
 	}
 
 	metrics.Server.CloseClient()
-	ctl.sessionCtx.ClientRegistry.MarkOfflineByRunID(ctl.runID)
+	ctl.clientRegistry.MarkOfflineByRunID(ctl.runID)
 	xl.Infof("client exit success")
 	close(ctl.doneCh)
 }
@@ -364,14 +381,14 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
 
 	content := &plugin.NewProxyContent{
 		User: plugin.UserInfo{
-			User:  ctl.sessionCtx.LoginMsg.User,
+			User:  ctl.loginMsg.User,
-			Metas: ctl.sessionCtx.LoginMsg.Metas,
+			Metas: ctl.loginMsg.Metas,
-			RunID: ctl.sessionCtx.LoginMsg.RunID,
+			RunID: ctl.loginMsg.RunID,
 		},
 		NewProxy: *inMsg,
 	}
 	var remoteAddr string
-	retContent, err := ctl.sessionCtx.PluginManager.NewProxy(content)
+	retContent, err := ctl.pluginManager.NewProxy(content)
 	if err == nil {
 		inMsg = &retContent.NewProxy
 		remoteAddr, err = ctl.RegisterProxy(inMsg)
@@ -384,15 +401,15 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
 	if err != nil {
 		xl.Warnf("new proxy [%s] type [%s] error: %v", inMsg.ProxyName, inMsg.ProxyType, err)
 		resp.Error = util.GenerateResponseErrorString(fmt.Sprintf("new proxy [%s] error", inMsg.ProxyName),
-			err, lo.FromPtr(ctl.sessionCtx.ServerCfg.DetailedErrorsToClient))
+			err, lo.FromPtr(ctl.serverCfg.DetailedErrorsToClient))
 	} else {
 		resp.RemoteAddr = remoteAddr
 		xl.Infof("new proxy [%s] type [%s] success", inMsg.ProxyName, inMsg.ProxyType)
-		clientID := ctl.sessionCtx.LoginMsg.ClientID
+		clientID := ctl.loginMsg.ClientID
 		if clientID == "" {
-			clientID = ctl.sessionCtx.LoginMsg.RunID
+			clientID = ctl.loginMsg.RunID
 		}
-		metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
+		metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.loginMsg.User, clientID)
 	}
 	_ = ctl.msgDispatcher.Send(resp)
 }
@@ -403,21 +420,21 @@ func (ctl *Control) handlePing(m msg.Message) {
 
 	content := &plugin.PingContent{
 		User: plugin.UserInfo{
-			User:  ctl.sessionCtx.LoginMsg.User,
+			User:  ctl.loginMsg.User,
-			Metas: ctl.sessionCtx.LoginMsg.Metas,
+			Metas: ctl.loginMsg.Metas,
-			RunID: ctl.sessionCtx.LoginMsg.RunID,
+			RunID: ctl.loginMsg.RunID,
 		},
 		Ping: *inMsg,
 	}
-	retContent, err := ctl.sessionCtx.PluginManager.Ping(content)
+	retContent, err := ctl.pluginManager.Ping(content)
 	if err == nil {
 		inMsg = &retContent.Ping
-		err = ctl.sessionCtx.AuthVerifier.VerifyPing(inMsg)
+		err = ctl.authVerifier.VerifyPing(inMsg)
 	}
 	if err != nil {
 		xl.Warnf("received invalid ping: %v", err)
 		_ = ctl.msgDispatcher.Send(&msg.Pong{
-			Error: util.GenerateResponseErrorString("invalid ping", err, lo.FromPtr(ctl.sessionCtx.ServerCfg.DetailedErrorsToClient)),
+			Error: util.GenerateResponseErrorString("invalid ping", err, lo.FromPtr(ctl.serverCfg.DetailedErrorsToClient)),
 		})
 		return
 	}
@@ -428,17 +445,17 @@ func (ctl *Control) handlePing(m msg.Message) {
 
 func (ctl *Control) handleNatHoleVisitor(m msg.Message) {
 	inMsg := m.(*msg.NatHoleVisitor)
-	ctl.sessionCtx.RC.NatHoleController.HandleVisitor(inMsg, ctl.msgTransporter, ctl.sessionCtx.LoginMsg.User)
+	ctl.rc.NatHoleController.HandleVisitor(inMsg, ctl.msgTransporter, ctl.loginMsg.User)
 }
 
 func (ctl *Control) handleNatHoleClient(m msg.Message) {
 	inMsg := m.(*msg.NatHoleClient)
-	ctl.sessionCtx.RC.NatHoleController.HandleClient(inMsg, ctl.msgTransporter)
+	ctl.rc.NatHoleController.HandleClient(inMsg, ctl.msgTransporter)
 }
 
 func (ctl *Control) handleNatHoleReport(m msg.Message) {
 	inMsg := m.(*msg.NatHoleReport)
-	ctl.sessionCtx.RC.NatHoleController.HandleReport(inMsg)
+	ctl.rc.NatHoleController.HandleReport(inMsg)
 }
 
 func (ctl *Control) handleCloseProxy(m msg.Message) {
@@ -451,15 +468,15 @@ func (ctl *Control) handleCloseProxy(m msg.Message) {
 func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err error) {
 	var pxyConf v1.ProxyConfigurer
 	// Load configures from NewProxy message and validate.
-	pxyConf, err = config.NewProxyConfigurerFromMsg(pxyMsg, ctl.sessionCtx.ServerCfg)
+	pxyConf, err = config.NewProxyConfigurerFromMsg(pxyMsg, ctl.serverCfg)
 	if err != nil {
 		return
 	}
 
 	// User info
 	userInfo := plugin.UserInfo{
-		User:  ctl.sessionCtx.LoginMsg.User,
+		User:  ctl.loginMsg.User,
-		Metas: ctl.sessionCtx.LoginMsg.Metas,
+		Metas: ctl.loginMsg.Metas,
 		RunID: ctl.runID,
 	}
 
@@ -467,22 +484,22 @@ func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err
 	// In fact, it creates different proxies based on the proxy type. We just call run() here.
 	pxy, err := proxy.NewProxy(ctl.ctx, &proxy.Options{
 		UserInfo:           userInfo,
-		LoginMsg:           ctl.sessionCtx.LoginMsg,
+		LoginMsg:           ctl.loginMsg,
 		PoolCount:          ctl.poolCount,
-		ResourceController: ctl.sessionCtx.RC,
+		ResourceController: ctl.rc,
 		GetWorkConnFn:      ctl.GetWorkConn,
 		Configurer:         pxyConf,
-		ServerCfg:          ctl.sessionCtx.ServerCfg,
+		ServerCfg:          ctl.serverCfg,
-		EncryptionKey:      ctl.sessionCtx.EncryptionKey,
+		EncryptionKey:      ctl.encryptionKey,
 	})
 	if err != nil {
 		return remoteAddr, err
 	}
 
 	// Check ports used number in each client
-	if ctl.sessionCtx.ServerCfg.MaxPortsPerClient > 0 {
+	if ctl.serverCfg.MaxPortsPerClient > 0 {
 		ctl.mu.Lock()
-		if ctl.portsUsedNum+pxy.GetUsedPortsNum() > int(ctl.sessionCtx.ServerCfg.MaxPortsPerClient) {
+		if ctl.portsUsedNum+pxy.GetUsedPortsNum() > int(ctl.serverCfg.MaxPortsPerClient) {
 			ctl.mu.Unlock()
 			err = fmt.Errorf("exceed the max_ports_per_client")
 			return
@@ -499,7 +516,7 @@ func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err
 		}()
 	}
 
-	if ctl.sessionCtx.PxyManager.Exist(pxyMsg.ProxyName) {
+	if ctl.pxyManager.Exist(pxyMsg.ProxyName) {
 		err = fmt.Errorf("proxy [%s] already exists", pxyMsg.ProxyName)
 		return
 	}
@@ -514,7 +531,7 @@ func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err
 		}
 	}()
 
-	err = ctl.sessionCtx.PxyManager.Add(pxyMsg.ProxyName, pxy)
+	err = ctl.pxyManager.Add(pxyMsg.ProxyName, pxy)
 	if err != nil {
 		return
 	}
@@ -533,11 +550,11 @@ func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
 		return
 	}
 
-	if ctl.sessionCtx.ServerCfg.MaxPortsPerClient > 0 {
+	if ctl.serverCfg.MaxPortsPerClient > 0 {
 		ctl.portsUsedNum -= pxy.GetUsedPortsNum()
 	}
 	pxy.Close()
-	ctl.sessionCtx.PxyManager.Del(pxy.GetName())
+	ctl.pxyManager.Del(pxy.GetName())
 	delete(ctl.proxies, closeMsg.ProxyName)
 	ctl.mu.Unlock()
 
@@ -545,16 +562,16 @@ func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
 
 	notifyContent := &plugin.CloseProxyContent{
 		User: plugin.UserInfo{
-			User:  ctl.sessionCtx.LoginMsg.User,
+			User:  ctl.loginMsg.User,
-			Metas: ctl.sessionCtx.LoginMsg.Metas,
+			Metas: ctl.loginMsg.Metas,
-			RunID: ctl.sessionCtx.LoginMsg.RunID,
+			RunID: ctl.loginMsg.RunID,
 		},
 		CloseProxy: msg.CloseProxy{
 			ProxyName: pxy.GetName(),
 		},
 	}
 	go func() {
-		_ = ctl.sessionCtx.PluginManager.CloseProxy(notifyContent)
+		_ = ctl.pluginManager.CloseProxy(notifyContent)
 	}()
 	return
 }

server/group/tcp.go

@@ -100,9 +100,8 @@ func (tg *TCPGroup) Listen(proxyName string, group string, groupKey string, addr
 		if err != nil {
 			return
 		}
-		tcpLn, errRet := net.Listen("tcp", net.JoinHostPort(addr, strconv.Itoa(realPort)))
+		tcpLn, errRet := net.Listen("tcp", net.JoinHostPort(addr, strconv.Itoa(port)))
 		if errRet != nil {
-			tg.ctl.portManager.Release(realPort)
 			err = errRet
 			return
 		}

server/proxy/http.go

@@ -75,13 +75,16 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
 		}
 	}()
 
-	domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
-
 	addrs := make([]string, 0)
-	for _, domain := range domains {
+	for _, domain := range pxy.cfg.CustomDomains {
+		if domain == "" {
+			continue
+		}
+
 		routeConfig.Domain = domain
 		for _, location := range locations {
 			routeConfig.Location = location
+
 			tmpRouteConfig := routeConfig
 
 			// handle group
@@ -90,6 +93,40 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
 				if err != nil {
 					return
 				}
+
+				pxy.closeFuncs = append(pxy.closeFuncs, func() {
+					pxy.rc.HTTPGroupCtl.UnRegister(pxy.name, pxy.cfg.LoadBalancer.Group, tmpRouteConfig)
+				})
+			} else {
+				// no group
+				err = pxy.rc.HTTPReverseProxy.Register(routeConfig)
+				if err != nil {
+					return
+				}
+				pxy.closeFuncs = append(pxy.closeFuncs, func() {
+					pxy.rc.HTTPReverseProxy.UnRegister(tmpRouteConfig)
+				})
+			}
+			addrs = append(addrs, util.CanonicalAddr(routeConfig.Domain, pxy.serverCfg.VhostHTTPPort))
+			xl.Infof("http proxy listen for host [%s] location [%s] group [%s], routeByHTTPUser [%s]",
+				routeConfig.Domain, routeConfig.Location, pxy.cfg.LoadBalancer.Group, pxy.cfg.RouteByHTTPUser)
+		}
+	}
+
+	if pxy.cfg.SubDomain != "" {
+		routeConfig.Domain = pxy.cfg.SubDomain + "." + pxy.serverCfg.SubDomainHost
+		for _, location := range locations {
+			routeConfig.Location = location
+
+			tmpRouteConfig := routeConfig
+
+			// handle group
+			if pxy.cfg.LoadBalancer.Group != "" {
+				err = pxy.rc.HTTPGroupCtl.Register(pxy.name, pxy.cfg.LoadBalancer.Group, pxy.cfg.LoadBalancer.GroupKey, routeConfig)
+				if err != nil {
+					return
+				}
+
 				pxy.closeFuncs = append(pxy.closeFuncs, func() {
 					pxy.rc.HTTPGroupCtl.UnRegister(pxy.name, pxy.cfg.LoadBalancer.Group, tmpRouteConfig)
 				})
@@ -102,7 +139,8 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
 					pxy.rc.HTTPReverseProxy.UnRegister(tmpRouteConfig)
 				})
 			}
-			addrs = append(addrs, util.CanonicalAddr(routeConfig.Domain, pxy.serverCfg.VhostHTTPPort))
+			addrs = append(addrs, util.CanonicalAddr(tmpRouteConfig.Domain, pxy.serverCfg.VhostHTTPPort))
+
 			xl.Infof("http proxy listen for host [%s] location [%s] group [%s], routeByHTTPUser [%s]",
 				routeConfig.Domain, routeConfig.Location, pxy.cfg.LoadBalancer.Group, pxy.cfg.RouteByHTTPUser)
 		}
@@ -130,7 +168,6 @@ func (pxy *HTTPProxy) GetRealConn(remoteAddr string) (workConn net.Conn, err err
 		rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
 		if err != nil {
 			xl.Errorf("create encryption stream error: %v", err)
-			tmpConn.Close()
 			return
 		}
 	}

server/proxy/https.go

@@ -53,10 +53,23 @@ func (pxy *HTTPSProxy) Run() (remoteAddr string, err error) {
 			pxy.Close()
 		}
 	}()
-	domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
-
 	addrs := make([]string, 0)
-	for _, domain := range domains {
+	for _, domain := range pxy.cfg.CustomDomains {
+		if domain == "" {
+			continue
+		}
+
+		l, err := pxy.listenForDomain(routeConfig, domain)
+		if err != nil {
+			return "", err
+		}
+		pxy.listeners = append(pxy.listeners, l)
+		addrs = append(addrs, util.CanonicalAddr(domain, pxy.serverCfg.VhostHTTPSPort))
+		xl.Infof("https proxy listen for host [%s] group [%s]", domain, pxy.cfg.LoadBalancer.Group)
+	}
+
+	if pxy.cfg.SubDomain != "" {
+		domain := pxy.cfg.SubDomain + "." + pxy.serverCfg.SubDomainHost
 		l, err := pxy.listenForDomain(routeConfig, domain)
 		if err != nil {
 			return "", err

server/proxy/proxy.go

@@ -150,7 +150,7 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
 			dstAddr, dstPortStr, _ = net.SplitHostPort(dst.String())
 			dstPort, _ = strconv.ParseUint(dstPortStr, 10, 16)
 		}
-		err = msg.WriteMsg(workConn, &msg.StartWorkConn{
+		err := msg.WriteMsg(workConn, &msg.StartWorkConn{
 			ProxyName: pxy.GetName(),
 			SrcAddr:   srcAddr,
 			SrcPort:   uint16(srcPort),
@@ -161,7 +161,6 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
 		if err != nil {
 			xl.Warnf("failed to send message to work connection from pool: %v, times: %d", err, i)
 			workConn.Close()
-			workConn = nil
 		} else {
 			break
 		}
@@ -174,36 +173,6 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
 	return
 }
 
-// startVisitorListener sets up a VisitorManager listener for visitor-based proxies (STCP, SUDP).
-func (pxy *BaseProxy) startVisitorListener(secretKey string, allowUsers []string, proxyType string) error {
-	// if allowUsers is empty, only allow same user from proxy
-	if len(allowUsers) == 0 {
-		allowUsers = []string{pxy.GetUserInfo().User}
-	}
-	listener, err := pxy.rc.VisitorManager.Listen(pxy.GetName(), secretKey, allowUsers)
-	if err != nil {
-		return err
-	}
-	pxy.listeners = append(pxy.listeners, listener)
-	pxy.xl.Infof("%s proxy custom listen success", proxyType)
-	pxy.startCommonTCPListenersHandler()
-	return nil
-}
-
-// buildDomains constructs a list of domains from custom domains and subdomain configuration.
-func (pxy *BaseProxy) buildDomains(customDomains []string, subDomain string) []string {
-	domains := make([]string, 0, len(customDomains)+1)
-	for _, d := range customDomains {
-		if d != "" {
-			domains = append(domains, d)
-		}
-	}
-	if subDomain != "" {
-		domains = append(domains, subDomain+"."+pxy.serverCfg.SubDomainHost)
-	}
-	return domains
-}
-
 // startCommonTCPListenersHandler start a goroutine handler for each listener.
 func (pxy *BaseProxy) startCommonTCPListenersHandler() {
 	xl := xlog.FromContextSafe(pxy.ctx)

server/proxy/stcp.go

@@ -41,7 +41,21 @@ func NewSTCPProxy(baseProxy *BaseProxy) Proxy {
 }
 
 func (pxy *STCPProxy) Run() (remoteAddr string, err error) {
-	err = pxy.startVisitorListener(pxy.cfg.Secretkey, pxy.cfg.AllowUsers, "stcp")
+	xl := pxy.xl
+	allowUsers := pxy.cfg.AllowUsers
+	// if allowUsers is empty, only allow same user from proxy
+	if len(allowUsers) == 0 {
+		allowUsers = []string{pxy.GetUserInfo().User}
+	}
+	listener, errRet := pxy.rc.VisitorManager.Listen(pxy.GetName(), pxy.cfg.Secretkey, allowUsers)
+	if errRet != nil {
+		err = errRet
+		return
+	}
+	pxy.listeners = append(pxy.listeners, listener)
+	xl.Infof("stcp proxy custom listen success")
+
+	pxy.startCommonTCPListenersHandler()
 	return
 }
 

server/proxy/sudp.go

@@ -41,7 +41,21 @@ func NewSUDPProxy(baseProxy *BaseProxy) Proxy {
 }
 
 func (pxy *SUDPProxy) Run() (remoteAddr string, err error) {
-	err = pxy.startVisitorListener(pxy.cfg.Secretkey, pxy.cfg.AllowUsers, "sudp")
+	xl := pxy.xl
+	allowUsers := pxy.cfg.AllowUsers
+	// if allowUsers is empty, only allow same user from proxy
+	if len(allowUsers) == 0 {
+		allowUsers = []string{pxy.GetUserInfo().User}
+	}
+	listener, errRet := pxy.rc.VisitorManager.Listen(pxy.GetName(), pxy.cfg.Secretkey, allowUsers)
+	if errRet != nil {
+		err = errRet
+		return
+	}
+	pxy.listeners = append(pxy.listeners, listener)
+	xl.Infof("sudp proxy custom listen success")
+
+	pxy.startCommonTCPListenersHandler()
 	return
 }
 

server/proxy/tcpmux.go

@@ -72,16 +72,26 @@ func (pxy *TCPMuxProxy) httpConnectListen(
 }
 
 func (pxy *TCPMuxProxy) httpConnectRun() (remoteAddr string, err error) {
-	domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
-
 	addrs := make([]string, 0)
-	for _, domain := range domains {
+	for _, domain := range pxy.cfg.CustomDomains {
+		if domain == "" {
+			continue
+		}
+
 		addrs, err = pxy.httpConnectListen(domain, pxy.cfg.RouteByHTTPUser, pxy.cfg.HTTPUser, pxy.cfg.HTTPPassword, addrs)
 		if err != nil {
 			return "", err
 		}
 	}
 
+	if pxy.cfg.SubDomain != "" {
+		addrs, err = pxy.httpConnectListen(pxy.cfg.SubDomain+"."+pxy.serverCfg.SubDomainHost,
+			pxy.cfg.RouteByHTTPUser, pxy.cfg.HTTPUser, pxy.cfg.HTTPPassword, addrs)
+		if err != nil {
+			return "", err
+		}
+	}
+
 	pxy.startCommonTCPListenersHandler()
 	remoteAddr = strings.Join(addrs, ",")
 	return remoteAddr, err

server/service.go

@@ -604,18 +604,8 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, inter
 		return err
 	}
 
-	ctl, err := NewControl(ctx, &SessionContext{
+	// TODO(fatedier): use SessionContext
-		RC:             svr.rc,
+	ctl, err := NewControl(ctx, svr.rc, svr.pxyManager, svr.pluginManager, authVerifier, svr.auth.EncryptionKey(), ctlConn, !internal, loginMsg, svr.cfg)
-		PxyManager:     svr.pxyManager,
-		PluginManager:  svr.pluginManager,
-		AuthVerifier:   authVerifier,
-		EncryptionKey:  svr.auth.EncryptionKey(),
-		Conn:           ctlConn,
-		ConnEncrypted:  !internal,
-		LoginMsg:       loginMsg,
-		ServerCfg:      svr.cfg,
-		ClientRegistry: svr.clientRegistry,
-	})
 	if err != nil {
 		xl.Warnf("create new controller error: %v", err)
 		// don't return detailed errors to client
@@ -636,6 +626,7 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, inter
 		ctl.Close()
 		return fmt.Errorf("client_id [%s] for user [%s] is already online", loginMsg.ClientID, loginMsg.User)
 	}
+	ctl.clientRegistry = svr.clientRegistry
 
 	ctl.Start()
 
@@ -661,9 +652,9 @@ func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn)
 	// server plugin hook
 	content := &plugin.NewWorkConnContent{
 		User: plugin.UserInfo{
-			User:  ctl.sessionCtx.LoginMsg.User,
+			User:  ctl.loginMsg.User,
-			Metas: ctl.sessionCtx.LoginMsg.Metas,
+			Metas: ctl.loginMsg.Metas,
-			RunID: ctl.sessionCtx.LoginMsg.RunID,
+			RunID: ctl.loginMsg.RunID,
 		},
 		NewWorkConn: *newMsg,
 	}
@@ -671,7 +662,7 @@ func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn)
 	if err == nil {
 		newMsg = &retContent.NewWorkConn
 		// Check auth.
-		err = ctl.sessionCtx.AuthVerifier.VerifyNewWorkConn(newMsg)
+		err = ctl.authVerifier.VerifyNewWorkConn(newMsg)
 	}
 	if err != nil {
 		xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
@@ -692,7 +683,7 @@ func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVis
 		if !exist {
 			return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
 		}
-		visitorUser = ctl.sessionCtx.LoginMsg.User
+		visitorUser = ctl.loginMsg.User
 	}
 	return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
 		newMsg.UseEncryption, newMsg.UseCompression, visitorUser)