Merge remote-tracking branch 'upstream/dev' into dev

2026-06-21 13:51:26 +08:00
parent d205f2bb35 9bde0b07de
commit d43a4e8950
352 changed files with 26713 additions and 13485 deletions
--- a/client/proxy/proxy.go
+++ b/client/proxy/proxy.go
@@ -16,6 +16,7 @@ package proxy

 import (
 	"context"
+	"fmt"
 	"io"
 	"net"
 	"reflect"
@@ -160,6 +161,33 @@ 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
 }
@@ -177,30 +205,14 @@ 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, err = libio.WithEncryption(remote, encKey)
-		if err != nil {
-			workConn.Close()
-			xl.Errorf("create encryption stream error: %v", err)
-			return
-		}
-	}
-	var compressionResourceRecycleFn func()
-	if baseCfg.Transport.UseCompression {
-		remote, compressionResourceRecycleFn = libio.WithCompressionFromPool(remote)
+
+	remote, recycleFn, err := pxy.wrapWorkConn(workConn, encKey)
+	if err != nil {
+		xl.Errorf("wrap work connection: %v", err)
+		return
 	}

 	// check if we need to send proxy protocol info
@@ -216,7 +228,6 @@ 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
 	}
@@ -225,12 +236,18 @@ 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),
@@ -247,6 +264,7 @@ 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
 		}
@@ -257,7 +275,4 @@ 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()
-	}
 }
--- a/client/proxy/proxy_wrapper.go
+++ b/client/proxy/proxy_wrapper.go
@@ -29,6 +29,7 @@ import (
 	"github.com/fatedier/frp/client/health"
 	v1 "github.com/fatedier/frp/pkg/config/v1"
 	"github.com/fatedier/frp/pkg/msg"
+	"github.com/fatedier/frp/pkg/naming"
 	"github.com/fatedier/frp/pkg/transport"
 	"github.com/fatedier/frp/pkg/util/xlog"
 	"github.com/fatedier/frp/pkg/vnet"
@@ -86,6 +87,8 @@ type Wrapper struct {

 	xl  *xlog.Logger
 	ctx context.Context
+
+	wireName string
 }

 func NewWrapper(
@@ -113,6 +116,7 @@ func NewWrapper(
 		vnetController: vnetController,
 		xl:             xl,
 		ctx:            xlog.NewContext(ctx, xl),
+		wireName:       naming.AddUserPrefix(clientCfg.User, baseInfo.Name),
 	}

 	if baseInfo.HealthCheck.Type != "" && baseInfo.LocalPort > 0 {
@@ -182,7 +186,7 @@ func (pw *Wrapper) Stop() {
 func (pw *Wrapper) close() {
 	_ = pw.handler(&event.CloseProxyPayload{
 		CloseProxyMsg: &msg.CloseProxy{
-			ProxyName: pw.Name,
+			ProxyName: pw.wireName,
 		},
 	})
 }
@@ -208,6 +212,7 @@ func (pw *Wrapper) checkWorker() {

 				var newProxyMsg msg.NewProxy
 				pw.Cfg.MarshalToMsg(&newProxyMsg)
+				newProxyMsg.ProxyName = pw.wireName
 				pw.lastSendStartMsg = now
 				_ = pw.handler(&event.StartProxyPayload{
 					NewProxyMsg: &newProxyMsg,
--- a/client/proxy/sudp.go
+++ b/client/proxy/sudp.go
@@ -17,7 +17,6 @@
 package proxy

 import (
-	"io"
 	"net"
 	"reflect"
 	"strconv"
@@ -25,17 +24,15 @@ 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"
 )

 func init() {
-	RegisterProxyFactory(reflect.TypeOf(&v1.SUDPProxyConfig{}), NewSUDPProxy)
+	RegisterProxyFactory(reflect.TypeFor[*v1.SUDPProxyConfig](), NewSUDPProxy)
 }

 type SUDPProxy struct {
@@ -83,27 +80,14 @@ 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())

-	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()
-		})
+	remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
+	if err != nil {
+		xl.Errorf("wrap work connection: %v", err)
+		return
 	}
-	if pxy.cfg.Transport.UseEncryption {
-		rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
-		if err != nil {
-			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 := conn
+	workConn := netpkg.WrapReadWriteCloserToConn(remote, conn)
+	payloadConn := msg.NewConn(workConn, msg.NewReadWriter(workConn, pxy.clientCfg.Transport.WireProtocol))
 	readCh := make(chan *msg.UDPPacket, 1024)
 	sendCh := make(chan msg.Message, 1024)
 	isClose := false
@@ -126,7 +110,7 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	}

 	// udp service <- frpc <- frps <- frpc visitor <- user
-	workConnReaderFn := func(conn net.Conn, readCh chan *msg.UDPPacket) {
+	workConnReaderFn := func(payloadConn *msg.Conn, readCh chan *msg.UDPPacket) {
 		defer closeFn()

 		for {
@@ -139,7 +123,7 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 			}

 			var udpMsg msg.UDPPacket
-			if errRet := msg.ReadMsgInto(conn, &udpMsg); errRet != nil {
+			if errRet := payloadConn.ReadMsgInto(&udpMsg); errRet != nil {
 				xl.Warnf("read from workConn for sudp error: %v", errRet)
 				return
 			}
@@ -154,7 +138,7 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	}

 	// udp service -> frpc -> frps -> frpc visitor -> user
-	workConnSenderFn := func(conn net.Conn, sendCh chan msg.Message) {
+	workConnSenderFn := func(payloadConn *msg.Conn, sendCh chan msg.Message) {
 		defer func() {
 			closeFn()
 			xl.Infof("writer goroutine for sudp work connection closed")
@@ -165,12 +149,12 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 			switch m := rawMsg.(type) {
 			case *msg.UDPPacket:
 				xl.Tracef("frpc send udp package to frpc visitor, [udp local: %v, remote: %v], [tcp work conn local: %v, remote: %v]",
-					m.LocalAddr.String(), m.RemoteAddr.String(), conn.LocalAddr().String(), conn.RemoteAddr().String())
+					m.LocalAddr.String(), m.RemoteAddr.String(), payloadConn.LocalAddr().String(), payloadConn.RemoteAddr().String())
 			case *msg.Ping:
 				xl.Tracef("frpc send ping message to frpc visitor")
 			}

-			if errRet = msg.WriteMsg(conn, rawMsg); errRet != nil {
+			if errRet = payloadConn.WriteMsg(rawMsg); errRet != nil {
 				xl.Errorf("sudp work write error: %v", errRet)
 				return
 			}
@@ -201,8 +185,8 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 		}
 	}

-	go workConnSenderFn(workConn, sendCh)
-	go workConnReaderFn(workConn, readCh)
+	go workConnSenderFn(payloadConn, sendCh)
+	go workConnReaderFn(payloadConn, readCh)
 	go heartbeatFn(sendCh)

 	udp.Forwarder(pxy.localAddr, readCh, sendCh, int(pxy.clientCfg.UDPPacketSize), pxy.cfg.Transport.ProxyProtocolVersion)
--- a/client/proxy/udp.go
+++ b/client/proxy/udp.go
@@ -17,24 +17,21 @@
 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"
 )

 func init() {
-	RegisterProxyFactory(reflect.TypeOf(&v1.UDPProxyConfig{}), NewUDPProxy)
+	RegisterProxyFactory(reflect.TypeFor[*v1.UDPProxyConfig](), NewUDPProxy)
 }

 type UDPProxy struct {
@@ -94,42 +91,30 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	// close resources related with old workConn
 	pxy.Close()

-	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()
-		})
+	remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
+	if err != nil {
+		xl.Errorf("wrap work connection: %v", err)
+		return
 	}
-	if pxy.cfg.Transport.UseEncryption {
-		rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
-		if err != nil {
-			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 = conn
+	pxy.workConn = netpkg.WrapReadWriteCloserToConn(remote, conn)
+	// Plain UDP payload follows the configured wire protocol for message framing.
+	payloadRW := msg.NewReadWriter(pxy.workConn, pxy.clientCfg.Transport.WireProtocol)
 	pxy.readCh = make(chan *msg.UDPPacket, 1024)
 	pxy.sendCh = make(chan msg.Message, 1024)
 	pxy.closed = false
 	pxy.mu.Unlock()

-	workConnReaderFn := func(conn net.Conn, readCh chan *msg.UDPPacket) {
+	workConnReaderFn := func(rw msg.ReadWriter, readCh chan *msg.UDPPacket) {
 		for {
 			var udpMsg msg.UDPPacket
-			if errRet := msg.ReadMsgInto(conn, &udpMsg); errRet != nil {
+			if errRet := rw.ReadMsgInto(&udpMsg); errRet != nil {
 				xl.Warnf("read from workConn for udp error: %v", errRet)
 				return
 			}
 			if errRet := errors.PanicToError(func() {
-				xl.Tracef("get udp package from workConn: %s", udpMsg.Content)
+				xl.Tracef("get udp package from workConn, len: %d", len(udpMsg.Content))
 				readCh <- &udpMsg
 			}); errRet != nil {
 				xl.Infof("reader goroutine for udp work connection closed: %v", errRet)
@@ -137,7 +122,7 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 			}
 		}
 	}
-	workConnSenderFn := func(conn net.Conn, sendCh chan msg.Message) {
+	workConnSenderFn := func(rw msg.ReadWriter, sendCh chan msg.Message) {
 		defer func() {
 			xl.Infof("writer goroutine for udp work connection closed")
 		}()
@@ -145,11 +130,11 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 		for rawMsg := range sendCh {
 			switch m := rawMsg.(type) {
 			case *msg.UDPPacket:
-				xl.Tracef("send udp package to workConn: %s", m.Content)
+				xl.Tracef("send udp package to workConn, len: %d", len(m.Content))
 			case *msg.Ping:
 				xl.Tracef("send ping message to udp workConn")
 			}
-			if errRet = msg.WriteMsg(conn, rawMsg); errRet != nil {
+			if errRet = rw.WriteMsg(rawMsg); errRet != nil {
 				xl.Errorf("udp work write error: %v", errRet)
 				return
 			}
@@ -168,8 +153,8 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 		}
 	}

-	go workConnSenderFn(pxy.workConn, pxy.sendCh)
-	go workConnReaderFn(pxy.workConn, pxy.readCh)
+	go workConnSenderFn(payloadRW, pxy.sendCh)
+	go workConnReaderFn(payloadRW, pxy.readCh)
 	go heartbeatFn(pxy.sendCh)

 	// Call Forwarder with proxy protocol version (empty string means no proxy protocol)
--- a/client/proxy/xtcp.go
+++ b/client/proxy/xtcp.go
@@ -27,13 +27,14 @@ import (

 	v1 "github.com/fatedier/frp/pkg/config/v1"
 	"github.com/fatedier/frp/pkg/msg"
+	"github.com/fatedier/frp/pkg/naming"
 	"github.com/fatedier/frp/pkg/nathole"
 	"github.com/fatedier/frp/pkg/transport"
 	netpkg "github.com/fatedier/frp/pkg/util/net"
 )

 func init() {
-	RegisterProxyFactory(reflect.TypeOf(&v1.XTCPProxyConfig{}), NewXTCPProxy)
+	RegisterProxyFactory(reflect.TypeFor[*v1.XTCPProxyConfig](), NewXTCPProxy)
 }

 type XTCPProxy struct {
@@ -56,8 +57,7 @@ func NewXTCPProxy(baseProxy *BaseProxy, cfg v1.ProxyConfigurer) Proxy {
 func (pxy *XTCPProxy) InWorkConn(conn net.Conn, startWorkConnMsg *msg.StartWorkConn) {
 	xl := pxy.xl
 	defer conn.Close()
-	var natHoleSidMsg msg.NatHoleSid
-	err := msg.ReadMsgInto(conn, &natHoleSidMsg)
+	natHoleSidMsg, err := readNatHoleSid(conn, pxy.clientCfg.Transport.WireProtocol)
 	if err != nil {
 		xl.Errorf("xtcp read from workConn error: %v", err)
 		return
@@ -85,7 +85,7 @@ func (pxy *XTCPProxy) InWorkConn(conn net.Conn, startWorkConnMsg *msg.StartWorkC
 	transactionID := nathole.NewTransactionID()
 	natHoleClientMsg := &msg.NatHoleClient{
 		TransactionID: transactionID,
-		ProxyName:     pxy.cfg.Name,
+		ProxyName:     naming.AddUserPrefix(pxy.clientCfg.User, pxy.cfg.Name),
 		Sid:           natHoleSidMsg.Sid,
 		MappedAddrs:   prepareResult.Addrs,
 		AssistedAddrs: prepareResult.AssistedAddrs,
@@ -130,6 +130,15 @@ func (pxy *XTCPProxy) InWorkConn(conn net.Conn, startWorkConnMsg *msg.StartWorkC
 	pxy.listenByQUIC(listenConn, raddr, startWorkConnMsg)
 }

+func readNatHoleSid(conn net.Conn, wireProtocol string) (*msg.NatHoleSid, error) {
+	workMsgConn := msg.NewConn(conn, msg.NewReadWriter(conn, wireProtocol))
+	var natHoleSidMsg msg.NatHoleSid
+	if err := workMsgConn.ReadMsgInto(&natHoleSidMsg); err != nil {
+		return nil, err
+	}
+	return &natHoleSidMsg, nil
+}
+
 func (pxy *XTCPProxy) listenByKCP(listenConn *net.UDPConn, raddr *net.UDPAddr, startWorkConnMsg *msg.StartWorkConn) {
 	xl := pxy.xl
 	listenConn.Close()
--- a/client/proxy/xtcp_test.go
+++ b/client/proxy/xtcp_test.go
@@ -0,0 +1,66 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !frps
+
+package proxy
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/fatedier/frp/pkg/msg"
+	"github.com/fatedier/frp/pkg/proto/wire"
+)
+
+func TestReadNatHoleSidUsesSelectedWireProtocol(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		wireProtocol string
+	}{
+		{name: "v2", wireProtocol: wire.ProtocolV2},
+		{name: "v1", wireProtocol: wire.ProtocolV1},
+		{name: "default", wireProtocol: ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client, server := net.Pipe()
+			defer client.Close()
+			defer server.Close()
+			setPipeDeadline(t, client, server)
+
+			errCh := make(chan error, 1)
+			go func() {
+				writer := msg.NewConn(server, msg.NewReadWriter(server, tc.wireProtocol))
+				errCh <- writer.WriteMsg(&msg.NatHoleSid{Sid: "sid"})
+			}()
+
+			out, err := readNatHoleSid(client, tc.wireProtocol)
+			require.NoError(t, err)
+			require.Equal(t, "sid", out.Sid)
+			require.NoError(t, <-errCh)
+		})
+	}
+}
+
+func setPipeDeadline(t *testing.T, conns ...net.Conn) {
+	t.Helper()
+
+	deadline := time.Now().Add(time.Second)
+	for _, conn := range conns {
+		require.NoError(t, conn.SetDeadline(deadline))
+	}
+}