diff --git a/Release.md b/Release.md index 26deaad0..0d2500d6 100644 --- a/Release.md +++ b/Release.md @@ -1,3 +1,7 @@ +## Features + +* UDP packet payloads for ordinary UDP proxies and SUDP now use a dedicated binary codec when frpc and frps successfully negotiate the capability under wire protocol v2, using a more compact wire representation. Wire protocol v1 remains JSON; wire protocol v2 falls back to JSON `UDPPacket` when the peer does not support or did not negotiate the capability. + ## Fixes * Fixed a server panic and remote denial of service caused by a client sending a negative `pool_count`. Negative values are now rejected before work-connection pool resources are allocated. diff --git a/client/control.go b/client/control.go index 5980ed34..4a354ff6 100644 --- a/client/control.go +++ b/client/control.go @@ -103,7 +103,7 @@ func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, erro sessionCtx.UDPPacketCodec, ) ctl.vm = visitor.NewManager(ctl.ctx, sessionCtx.RunID, sessionCtx.Common, - ctl.connectServer, ctl.msgTransporter, sessionCtx.VnetController) + ctl.connectServer, ctl.msgTransporter, sessionCtx.VnetController, sessionCtx.UDPPacketCodec) return ctl, nil } diff --git a/client/proxy/sudp.go b/client/proxy/sudp.go index 7db6d991..b5613856 100644 --- a/client/proxy/sudp.go +++ b/client/proxy/sudp.go @@ -87,7 +87,13 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) { } workConn := netpkg.WrapReadWriteCloserToConn(remote, conn) - payloadConn := msg.NewConn(workConn, msg.NewReadWriter(workConn, pxy.clientCfg.Transport.WireProtocol)) + payloadRW, err := msg.NewUDPPacketReadWriter(workConn, pxy.clientCfg.Transport.WireProtocol, pxy.udpPacketCodec) + if err != nil { + xl.Errorf("create SUDP packet read writer: %v", err) + _ = workConn.Close() + return + } + payloadConn := msg.NewConn(workConn, payloadRW) readCh := make(chan *msg.UDPPacket, 1024) sendCh := make(chan msg.Message, 1024) isClose := false diff --git a/client/service_shutdown_test.go b/client/service_shutdown_test.go index 3b956726..73c7bd3d 100644 --- a/client/service_shutdown_test.go +++ b/client/service_shutdown_test.go @@ -35,7 +35,7 @@ func newGracefulCloseTestService() *Service { doneCh: make(chan struct{}), } ctl.pm = proxy.NewManager(ctx, common, nil, nil, nil, "") - ctl.vm = visitor.NewManager(ctx, "graceful-close-race", common, nil, nil, nil) + ctl.vm = visitor.NewManager(ctx, "graceful-close-race", common, nil, nil, nil, "") return &Service{ctl: ctl, cancel: context.CancelCauseFunc(func(error) {})} } diff --git a/client/visitor/sudp.go b/client/visitor/sudp.go index 91d57da6..f1939bda 100644 --- a/client/visitor/sudp.go +++ b/client/visitor/sudp.go @@ -113,7 +113,13 @@ func (sv *SUDPVisitor) dispatcher() { func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) { xl := xlog.FromContextSafe(sv.ctx) xl.Debugf("starting sudp proxy worker") - payloadConn := msg.NewConn(workConn, msg.NewReadWriter(workConn, sv.clientCfg.Transport.WireProtocol)) + payloadRW, err := msg.NewUDPPacketReadWriter(workConn, sv.clientCfg.Transport.WireProtocol, udpPacketCodecFromHelper(sv.helper)) + if err != nil { + xl.Errorf("create SUDP packet read writer: %v", err) + _ = workConn.Close() + return + } + payloadConn := msg.NewConn(workConn, payloadRW) wg := &sync.WaitGroup{} wg.Add(2) diff --git a/client/visitor/visitor.go b/client/visitor/visitor.go index dff0bb94..eccad013 100644 --- a/client/visitor/visitor.go +++ b/client/visitor/visitor.go @@ -50,6 +50,17 @@ type Helper interface { RunID() string } +type udpPacketCodecProvider interface { + UDPPacketCodec() string +} + +func udpPacketCodecFromHelper(helper Helper) string { + if provider, ok := helper.(udpPacketCodecProvider); ok { + return provider.UDPPacketCodec() + } + return "" +} + // Visitor is used for forward traffics from local port tot remote service. type Visitor interface { Run() error diff --git a/client/visitor/visitor_manager.go b/client/visitor/visitor_manager.go index 1ca194bd..3e614f7a 100644 --- a/client/visitor/visitor_manager.go +++ b/client/visitor/visitor_manager.go @@ -53,7 +53,12 @@ func NewManager( connectServer func() (*msg.Conn, error), msgTransporter transport.MessageTransporter, vnetController *vnet.Controller, + udpPacketCodecs ...string, ) *Manager { + udpPacketCodec := "" + if len(udpPacketCodecs) > 0 { + udpPacketCodec = udpPacketCodecs[0] + } m := &Manager{ clientCfg: clientCfg, cfgs: make(map[string]v1.VisitorConfigurer), @@ -68,6 +73,7 @@ func NewManager( vnetController: vnetController, transferConnFn: m.TransferConn, runID: runID, + udpPacketCodec: udpPacketCodec, } return m } @@ -205,6 +211,7 @@ type visitorHelperImpl struct { vnetController *vnet.Controller transferConnFn func(name string, conn net.Conn) error runID string + udpPacketCodec string } func (v *visitorHelperImpl) ConnectServer() (*msg.Conn, error) { @@ -226,3 +233,7 @@ func (v *visitorHelperImpl) VNetController() *vnet.Controller { func (v *visitorHelperImpl) RunID() string { return v.runID } + +func (v *visitorHelperImpl) UDPPacketCodec() string { + return v.udpPacketCodec +} diff --git a/pkg/msg/udp_benchmark_test.go b/pkg/msg/udp_benchmark_test.go new file mode 100644 index 00000000..575938ec --- /dev/null +++ b/pkg/msg/udp_benchmark_test.go @@ -0,0 +1,199 @@ +// 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. + +package msg + +import ( + "bytes" + "fmt" + "net" + "runtime" + "testing" + + "github.com/fatedier/frp/pkg/proto/wire" +) + +type udpBenchmarkCase struct { + name string + packet *UDPPacket +} + +var ( + udpBenchmarkBytesSink []byte + udpBenchmarkMessageSink Message +) + +func udpBenchmarkCases(payloadSize int) []udpBenchmarkCase { + content := bytes.Repeat([]byte{0x5a}, payloadSize) + return []udpBenchmarkCase{ + { + name: "ipv4-remote", + packet: &UDPPacket{ + Content: content, + RemoteAddr: &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 12345}, + }, + }, + { + name: "ipv4-local-remote", + packet: &UDPPacket{ + Content: content, + LocalAddr: &net.UDPAddr{IP: net.ParseIP("192.0.2.2"), Port: 23456}, + RemoteAddr: &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 12345}, + }, + }, + { + name: "ipv6-remote", + packet: &UDPPacket{ + Content: content, + RemoteAddr: &net.UDPAddr{IP: net.ParseIP("2001:db8::1"), Port: 12345}, + }, + }, + { + name: "ipv6-local-remote", + packet: &UDPPacket{ + Content: content, + LocalAddr: &net.UDPAddr{IP: net.ParseIP("2001:db8::2"), Port: 23456, Zone: "bench0"}, + RemoteAddr: &net.UDPAddr{IP: net.ParseIP("2001:db8::1"), Port: 12345, Zone: "bench1"}, + }, + }, + } +} + +func TestUDPPacketV2FrameSizes(t *testing.T) { + t.Logf("environment go=%s goos=%s goarch=%s gomaxprocs=%d", runtime.Version(), runtime.GOOS, runtime.GOARCH, runtime.GOMAXPROCS(0)) + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range udpBenchmarkCases(payloadSize) { + jsonFrame := udpBenchmarkWireBytes(t, tc.packet, "") + binaryFrame := udpBenchmarkWireBytes(t, tc.packet, wire.UDPPacketCodecBinary) + saving := 100 * float64(len(jsonFrame)-len(binaryFrame)) / float64(len(jsonFrame)) + t.Logf("frame payload=%d case=%s json_bytes=%d binary_bytes=%d binary_saving_pct=%.2f", payloadSize, tc.name, len(jsonFrame), len(binaryFrame), saving) + } + } +} + +func udpBenchmarkWireBytes(t testing.TB, packet *UDPPacket, codec string) []byte { + t.Helper() + var buf bytes.Buffer + rw, err := NewUDPPacketReadWriter(&buf, wire.ProtocolV2, codec) + if err != nil { + t.Fatalf("create UDP read writer: %v", err) + } + if err := rw.WriteMsg(packet); err != nil { + t.Fatalf("write UDP packet: %v", err) + } + return append([]byte(nil), buf.Bytes()...) +} + +type udpBenchmarkReadWriter struct { + reader bytes.Reader +} + +func (rw *udpBenchmarkReadWriter) Read(p []byte) (int, error) { + return rw.reader.Read(p) +} + +func (rw *udpBenchmarkReadWriter) Write(p []byte) (int, error) { + return len(p), nil +} + +func (rw *udpBenchmarkReadWriter) Reset(p []byte) { + rw.reader.Reset(p) +} + +func udpBenchmarkValidatePacket(b testing.TB, got, want *UDPPacket) { + b.Helper() + if !bytes.Equal(got.Content, want.Content) || !udpBenchmarkUDPAddrEqual(got.LocalAddr, want.LocalAddr) || + !udpBenchmarkUDPAddrEqual(got.RemoteAddr, want.RemoteAddr) { + b.Fatalf("decoded packet mismatch: got %+v, want %+v", got, want) + } +} + +func udpBenchmarkUDPAddrEqual(got, want *net.UDPAddr) bool { + if got == nil || want == nil { + return got == want + } + return got.IP.Equal(want.IP) && got.Port == want.Port && got.Zone == want.Zone +} + +func BenchmarkUDPPacketV2CodecWrite(b *testing.B) { + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range udpBenchmarkCases(payloadSize) { + for _, codec := range []struct { + name string + value string + }{ + {name: "json", value: ""}, + {name: "binary", value: wire.UDPPacketCodecBinary}, + } { + b.Run(fmt.Sprintf("payload-%d/%s/%s", payloadSize, tc.name, codec.name), func(b *testing.B) { + var buf bytes.Buffer + rw, err := NewUDPPacketReadWriter(&buf, wire.ProtocolV2, codec.value) + if err != nil { + b.Fatal(err) + } + expected := udpBenchmarkWireBytes(b, tc.packet, codec.value) + b.SetBytes(int64(len(expected))) + for b.Loop() { + buf.Reset() + if err := rw.WriteMsg(tc.packet); err != nil { + b.Fatal(err) + } + } + if !bytes.Equal(buf.Bytes(), expected) { + b.Fatalf("encoded packet mismatch: got %d bytes, want %d", buf.Len(), len(expected)) + } + udpBenchmarkBytesSink = buf.Bytes() + }) + } + } + } +} + +func BenchmarkUDPPacketV2CodecRead(b *testing.B) { + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range udpBenchmarkCases(payloadSize) { + for _, codec := range []struct { + name string + value string + }{ + {name: "json", value: ""}, + {name: "binary", value: wire.UDPPacketCodecBinary}, + } { + b.Run(fmt.Sprintf("payload-%d/%s/%s", payloadSize, tc.name, codec.name), func(b *testing.B) { + encoded := udpBenchmarkWireBytes(b, tc.packet, codec.value) + stream := &udpBenchmarkReadWriter{} + rw, err := NewUDPPacketReadWriter(stream, wire.ProtocolV2, codec.value) + if err != nil { + b.Fatal(err) + } + var decoded Message + b.SetBytes(int64(len(encoded))) + for b.Loop() { + stream.Reset(encoded) + decoded, err = rw.ReadMsg() + if err != nil { + b.Fatal(err) + } + } + packet, ok := decoded.(*UDPPacket) + if !ok { + b.Fatalf("decoded message type %T, want *UDPPacket", decoded) + } + udpBenchmarkValidatePacket(b, packet, tc.packet) + udpBenchmarkMessageSink = decoded + }) + } + } + } +} diff --git a/server/control.go b/server/control.go index 1690b01b..8009d8ac 100644 --- a/server/control.go +++ b/server/control.go @@ -286,7 +286,7 @@ func (cm *ControlManager) GetByID(runID string) (ctl *Control, ok bool) { // admitVisitorByRunID commits a visitor admission against the current running // control while its run and lifecycle ownership are held. The callback must // only perform the in-memory, buffered visitor admission. -func (cm *ControlManager) admitVisitorByRunID(runID string, admit func(user string) error) (bool, error) { +func (cm *ControlManager) admitVisitorByRunID(runID string, admit func(user, wireProtocol, udpPacketCodec string) error) (bool, error) { entry, ok := cm.lockCurrentRun(runID, false) if !ok { return false, nil @@ -299,7 +299,7 @@ func (cm *ControlManager) admitVisitorByRunID(runID string, admit func(user stri if ctl.state != controlStateRunning { return false, nil } - return true, admit(ctl.sessionCtx.LoginMsg.User) + return true, admit(ctl.sessionCtx.LoginMsg.User, ctl.sessionCtx.WireProtocol, ctl.sessionCtx.UDPPacketCodec) } // RegisterWorkConn transfers conn to ctl only if ctl is still the current diff --git a/server/proxy/proxy.go b/server/proxy/proxy.go index 046794b6..e0dbb7f2 100644 --- a/server/proxy/proxy.go +++ b/server/proxy/proxy.go @@ -328,10 +328,18 @@ func (pxy *BaseProxy) handleUserTCPConnection(userConn net.Conn) { func (pxy *BaseProxy) joinUserConnection(local io.ReadWriteCloser, userConn net.Conn, proxyType string, xl *xlog.Logger) (int64, int64, []error) { visitorWireProtocol := wireProtocolFromConn(userConn) - if proxyType == string(v1.ProxyTypeSUDP) && isMixedWireProtocol(pxy.wireProtocol, visitorWireProtocol) { - xl.Infof("bridge mixed SUDP payload codecs, proxy wireProtocol [%s], visitor wireProtocol [%s]", - normalizeWireProtocol(pxy.wireProtocol), normalizeWireProtocol(visitorWireProtocol)) - return joinSUDPMessageBridge(local, userConn, pxy.wireProtocol, visitorWireProtocol, xl) + visitorUDPPacketCodec := udpPacketCodecFromConn(userConn) + if proxyType == string(v1.ProxyTypeSUDP) { + mixed, err := isMixedSUDPPacketEncoding(pxy.wireProtocol, pxy.udpPacketCodec, visitorWireProtocol, visitorUDPPacketCodec) + if err != nil { + return 0, 0, []error{err} + } + if mixed { + xl.Infof("bridge mixed SUDP payload codecs, proxy [%s/%s], visitor [%s/%s]", + normalizeWireProtocol(pxy.wireProtocol), pxy.udpPacketCodec, + normalizeWireProtocol(visitorWireProtocol), visitorUDPPacketCodec) + return joinSUDPMessageBridge(local, userConn, pxy.wireProtocol, pxy.udpPacketCodec, visitorWireProtocol, visitorUDPPacketCodec, xl) + } } return libio.Join(local, userConn) } @@ -340,6 +348,10 @@ type wireProtocolGetter interface { WireProtocol() string } +type udpPacketCodecGetter interface { + UDPPacketCodec() string +} + func wireProtocolFromConn(conn net.Conn) string { if getter, ok := conn.(wireProtocolGetter); ok { return getter.WireProtocol() @@ -347,10 +359,46 @@ func wireProtocolFromConn(conn net.Conn) string { return "" } +func udpPacketCodecFromConn(conn net.Conn) string { + if getter, ok := conn.(udpPacketCodecGetter); ok { + return getter.UDPPacketCodec() + } + return "" +} + func isMixedWireProtocol(left, right string) bool { return normalizeWireProtocol(left) != normalizeWireProtocol(right) } +func isMixedSUDPPacketEncoding(leftWire, leftCodec, rightWire, rightCodec string) (bool, error) { + leftCodec, err := normalizeUDPPacketCodec(leftWire, leftCodec) + if err != nil { + return false, fmt.Errorf("invalid left SUDP packet encoding: %w", err) + } + rightCodec, err = normalizeUDPPacketCodec(rightWire, rightCodec) + if err != nil { + return false, fmt.Errorf("invalid right SUDP packet encoding: %w", err) + } + return normalizeWireProtocol(leftWire) != normalizeWireProtocol(rightWire) || leftCodec != rightCodec, nil +} + +func normalizeUDPPacketCodec(wireProtocol, codec string) (string, error) { + switch wireProtocol { + case "", wire.ProtocolV1: + if codec != "" { + return "", fmt.Errorf("UDP packet codec %q requires wire protocol v2", codec) + } + return "", nil + case wire.ProtocolV2: + if codec == "" || codec == wire.UDPPacketCodecBinary { + return codec, nil + } + return "", fmt.Errorf("unsupported UDP packet codec %q", codec) + default: + return "", fmt.Errorf("unsupported wire protocol %q", wireProtocol) + } +} + func normalizeWireProtocol(wireProtocol string) string { if wireProtocol == wire.ProtocolV2 { return wire.ProtocolV2 @@ -362,13 +410,21 @@ func joinSUDPMessageBridge( proxyConn io.ReadWriteCloser, visitorConn io.ReadWriteCloser, proxyWireProtocol string, + proxyUDPPacketCodec string, visitorWireProtocol string, + visitorUDPPacketCodec string, xl *xlog.Logger, ) (inCount int64, outCount int64, errs []error) { // The mixed bridge decodes and re-encodes messages, so raw framed byte counts // are not available. Count UDP payload bytes and ignore heartbeat traffic. - proxyRW := msg.NewReadWriter(proxyConn, proxyWireProtocol) - visitorRW := msg.NewReadWriter(visitorConn, visitorWireProtocol) + proxyRW, err := msg.NewUDPPacketReadWriter(proxyConn, proxyWireProtocol, proxyUDPPacketCodec) + if err != nil { + return 0, 0, []error{err} + } + visitorRW, err := msg.NewUDPPacketReadWriter(visitorConn, visitorWireProtocol, visitorUDPPacketCodec) + if err != nil { + return 0, 0, []error{err} + } var ( once sync.Once diff --git a/server/proxy/sudp_benchmark_test.go b/server/proxy/sudp_benchmark_test.go new file mode 100644 index 00000000..a2e337d9 --- /dev/null +++ b/server/proxy/sudp_benchmark_test.go @@ -0,0 +1,232 @@ +// 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. + +package proxy + +import ( + "bytes" + "fmt" + "io" + "net" + "testing" + + "github.com/fatedier/frp/pkg/msg" + "github.com/fatedier/frp/pkg/proto/wire" +) + +type sudpPathBenchmarkCase struct { + name string + packet *msg.UDPPacket +} + +var sudpPathBenchmarkBytesSink []byte + +func sudpPathBenchmarkCases(payloadSize int) []sudpPathBenchmarkCase { + content := bytes.Repeat([]byte{0x5a}, payloadSize) + return []sudpPathBenchmarkCase{ + { + name: "ipv4-remote", + packet: &msg.UDPPacket{ + Content: content, + RemoteAddr: &net.UDPAddr{ + IP: net.ParseIP("192.0.2.1"), Port: 12345, + }, + }, + }, + { + name: "ipv4-local-remote", + packet: &msg.UDPPacket{ + Content: content, + LocalAddr: &net.UDPAddr{ + IP: net.ParseIP("192.0.2.2"), Port: 23456, + }, + RemoteAddr: &net.UDPAddr{ + IP: net.ParseIP("192.0.2.1"), Port: 12345, + }, + }, + }, + { + name: "ipv6-remote", + packet: &msg.UDPPacket{ + Content: content, + RemoteAddr: &net.UDPAddr{ + IP: net.ParseIP("2001:db8::1"), Port: 12345, + }, + }, + }, + { + name: "ipv6-local-remote", + packet: &msg.UDPPacket{ + Content: content, + LocalAddr: &net.UDPAddr{ + IP: net.ParseIP("2001:db8::2"), Port: 23456, Zone: "bench0", + }, + RemoteAddr: &net.UDPAddr{ + IP: net.ParseIP("2001:db8::1"), Port: 12345, Zone: "bench1", + }, + }, + }, + } +} + +type sudpPathReadWriter struct { + reader bytes.Reader +} + +func (rw *sudpPathReadWriter) Read(p []byte) (int, error) { return rw.reader.Read(p) } +func (rw *sudpPathReadWriter) Write(p []byte) (int, error) { return len(p), nil } +func (rw *sudpPathReadWriter) Reset(p []byte) { rw.reader.Reset(p) } + +func sudpPathWireBytes(b testing.TB, packet *msg.UDPPacket, codec string) []byte { + b.Helper() + var buf bytes.Buffer + rw, err := msg.NewUDPPacketReadWriter(&buf, wire.ProtocolV2, codec) + if err != nil { + b.Fatal(err) + } + if err := rw.WriteMsg(packet); err != nil { + b.Fatal(err) + } + return append([]byte(nil), buf.Bytes()...) +} + +func sudpPathCopyFrame(dst, src []byte) []byte { + copy(dst, src) + return dst +} + +func BenchmarkSUDPInMemoryFrameCopy(b *testing.B) { + // This is an in-memory copy of an already encoded frame. It is a proxy for + // frame-size-dependent copy work, not a benchmark of libio.Join or sockets. + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range sudpPathBenchmarkCases(payloadSize) { + for _, codec := range []struct { + name string + value string + }{ + {name: "json", value: ""}, + {name: "binary", value: wire.UDPPacketCodecBinary}, + } { + b.Run(fmt.Sprintf("payload-%d/%s/%s", payloadSize, tc.name, codec.name), func(b *testing.B) { + encoded := sudpPathWireBytes(b, tc.packet, codec.value) + dst := make([]byte, len(encoded)) + b.SetBytes(int64(len(encoded))) + for b.Loop() { + dst = sudpPathCopyFrame(dst, encoded) + } + if !bytes.Equal(dst, encoded) { + b.Fatal("copied frame does not match source") + } + sudpPathBenchmarkBytesSink = dst + }) + } + } + } +} + +func BenchmarkSUDPEndpointCodecPair(b *testing.B) { + // This measures an in-memory decode and re-encode with the same codec. It + // does not include the live SUDP server path, sockets, goroutines, or I/O. + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range sudpPathBenchmarkCases(payloadSize) { + for _, codec := range []struct { + name string + value string + }{ + {name: "json", value: ""}, + {name: "binary", value: wire.UDPPacketCodecBinary}, + } { + b.Run(fmt.Sprintf("payload-%d/%s/%s", payloadSize, tc.name, codec.name), func(b *testing.B) { + encoded := sudpPathWireBytes(b, tc.packet, codec.value) + from := &sudpPathReadWriter{} + to := &bytes.Buffer{} + fromRW, err := msg.NewUDPPacketReadWriter(from, wire.ProtocolV2, codec.value) + if err != nil { + b.Fatal(err) + } + toRW, err := msg.NewUDPPacketReadWriter(to, wire.ProtocolV2, codec.value) + if err != nil { + b.Fatal(err) + } + b.SetBytes(int64(len(encoded))) + for b.Loop() { + from.Reset(encoded) + to.Reset() + m, err := fromRW.ReadMsg() + if err != nil { + b.Fatal(err) + } + if err := toRW.WriteMsg(m); err != nil { + b.Fatal(err) + } + } + if !bytes.Equal(to.Bytes(), encoded) { + b.Fatalf("re-encoded packet mismatch: got %d bytes, want %d", to.Len(), len(encoded)) + } + sudpPathBenchmarkBytesSink = to.Bytes() + }) + } + } + } +} + +func BenchmarkSUDPMixedCodecTranscodeModel(b *testing.B) { + // This exercises the real codec decode/re-encode pair used by the mixed + // bridge, excluding sockets, goroutines, crypto, compression, and framing I/O. + for _, payloadSize := range []int{64, 512, 1200, 1472} { + for _, tc := range sudpPathBenchmarkCases(payloadSize) { + for _, direction := range []struct { + name string + from string + to string + }{ + {name: "json-to-binary", from: "", to: wire.UDPPacketCodecBinary}, + {name: "binary-to-json", from: wire.UDPPacketCodecBinary, to: ""}, + } { + b.Run(fmt.Sprintf("payload-%d/%s/%s", payloadSize, tc.name, direction.name), func(b *testing.B) { + encoded := sudpPathWireBytes(b, tc.packet, direction.from) + expected := sudpPathWireBytes(b, tc.packet, direction.to) + from := &sudpPathReadWriter{} + to := &bytes.Buffer{} + fromRW, err := msg.NewUDPPacketReadWriter(from, wire.ProtocolV2, direction.from) + if err != nil { + b.Fatal(err) + } + toRW, err := msg.NewUDPPacketReadWriter(to, wire.ProtocolV2, direction.to) + if err != nil { + b.Fatal(err) + } + b.SetBytes(int64(len(encoded))) + for b.Loop() { + from.Reset(encoded) + to.Reset() + m, err := fromRW.ReadMsg() + if err != nil { + b.Fatal(err) + } + if err := toRW.WriteMsg(m); err != nil { + b.Fatal(err) + } + } + if !bytes.Equal(to.Bytes(), expected) { + b.Fatalf("transcoded packet mismatch: got %d bytes, want %d", to.Len(), len(expected)) + } + sudpPathBenchmarkBytesSink = to.Bytes() + }) + } + } + } +} + +var _ io.ReadWriter = (*sudpPathReadWriter)(nil) diff --git a/server/proxy/sudp_test.go b/server/proxy/sudp_test.go index e2986eed..b7ea9bc6 100644 --- a/server/proxy/sudp_test.go +++ b/server/proxy/sudp_test.go @@ -18,22 +18,27 @@ import ( "bufio" "bytes" "encoding/binary" + "io" + "net" "testing" + "time" "github.com/stretchr/testify/require" + v1 "github.com/fatedier/frp/pkg/config/v1" "github.com/fatedier/frp/pkg/msg" "github.com/fatedier/frp/pkg/proto/wire" + "github.com/fatedier/frp/pkg/util/xlog" ) func TestSUDPBridgeTranscodesProxyV1ToVisitorV2(t *testing.T) { var in, out bytes.Buffer - writeSUDPBridgeMsg(t, &in, wire.ProtocolV1, &msg.UDPPacket{Content: []byte("proxy-to-visitor")}) + writeSUDPBridgeMsg(t, &in, wire.ProtocolV1, "", &msg.UDPPacket{Content: []byte("proxy-to-visitor")}) var count int64 err := bridgeSUDPProxyToVisitor( - msg.NewReadWriter(&in, wire.ProtocolV1), - msg.NewReadWriter(&out, wire.ProtocolV2), + newSUDPBridgeRW(t, &in, wire.ProtocolV1, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV2, ""), &count, nil, ) @@ -53,12 +58,12 @@ func TestSUDPBridgeTranscodesProxyV1ToVisitorV2(t *testing.T) { func TestSUDPBridgeTranscodesVisitorV2ToProxyV1(t *testing.T) { var in, out bytes.Buffer - writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, &msg.UDPPacket{Content: []byte("visitor-to-proxy")}) + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, "", &msg.UDPPacket{Content: []byte("visitor-to-proxy")}) var count int64 err := bridgeSUDPVisitorToProxy( - msg.NewReadWriter(&in, wire.ProtocolV2), - msg.NewReadWriter(&out, wire.ProtocolV1), + newSUDPBridgeRW(t, &in, wire.ProtocolV2, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV1, ""), &count, nil, ) @@ -76,33 +81,67 @@ func TestSUDPBridgeTranscodesVisitorV2ToProxyV1(t *testing.T) { require.Equal(t, []byte("visitor-to-proxy"), got.Content) } -func TestSUDPBridgeForwardsProxyPing(t *testing.T) { +func TestSUDPBridgeTranscodesProxyV2BinaryToVisitorV2JSON(t *testing.T) { var in, out bytes.Buffer - writeSUDPBridgeMsg(t, &in, wire.ProtocolV1, &msg.Ping{}) + packet := newSUDPBridgeUDPPacket("proxy-binary-to-json") + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, wire.UDPPacketCodecBinary, packet) var count int64 err := bridgeSUDPProxyToVisitor( - msg.NewReadWriter(&in, wire.ProtocolV1), - msg.NewReadWriter(&out, wire.ProtocolV2), + newSUDPBridgeRW(t, &in, wire.ProtocolV2, wire.UDPPacketCodecBinary), + newSUDPBridgeRW(t, &out, wire.ProtocolV2, ""), + &count, + nil, + ) + require.NoError(t, err) + require.Equal(t, int64(len(packet.Content)), count) + requireV2UDPPacketFrame(t, &out, msg.V2TypeUDPPacket, packet) +} + +func TestSUDPBridgeTranscodesVisitorV2JSONToProxyV2Binary(t *testing.T) { + var in, out bytes.Buffer + packet := newSUDPBridgeUDPPacket("visitor-json-to-binary") + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, "", packet) + + var count int64 + err := bridgeSUDPVisitorToProxy( + newSUDPBridgeRW(t, &in, wire.ProtocolV2, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV2, wire.UDPPacketCodecBinary), + &count, + nil, + ) + require.NoError(t, err) + require.Equal(t, int64(len(packet.Content)), count) + requireV2UDPPacketFrame(t, &out, msg.V2TypeUDPPacketBinary, packet) +} + +func TestSUDPBridgeForwardsProxyPing(t *testing.T) { + var in, out bytes.Buffer + writeSUDPBridgeMsg(t, &in, wire.ProtocolV1, "", &msg.Ping{}) + + var count int64 + err := bridgeSUDPProxyToVisitor( + newSUDPBridgeRW(t, &in, wire.ProtocolV1, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV2, ""), &count, nil, ) require.NoError(t, err) require.Zero(t, count) - rawMsg, err := msg.NewReadWriter(&out, wire.ProtocolV2).ReadMsg() + rawMsg, err := newSUDPBridgeRW(t, &out, wire.ProtocolV2, "").ReadMsg() require.NoError(t, err) require.IsType(t, &msg.Ping{}, rawMsg) } func TestSUDPBridgeDropsVisitorPing(t *testing.T) { var in, out bytes.Buffer - writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, &msg.Ping{}) + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, "", &msg.Ping{}) var count int64 err := bridgeSUDPVisitorToProxy( - msg.NewReadWriter(&in, wire.ProtocolV2), - msg.NewReadWriter(&out, wire.ProtocolV1), + newSUDPBridgeRW(t, &in, wire.ProtocolV2, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV1, ""), &count, nil, ) @@ -113,12 +152,12 @@ func TestSUDPBridgeDropsVisitorPing(t *testing.T) { func TestSUDPBridgeRejectsUnknownVisitorMessage(t *testing.T) { var in, out bytes.Buffer - writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, &msg.Pong{}) + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, "", &msg.Pong{}) var count int64 err := bridgeSUDPVisitorToProxy( - msg.NewReadWriter(&in, wire.ProtocolV2), - msg.NewReadWriter(&out, wire.ProtocolV1), + newSUDPBridgeRW(t, &in, wire.ProtocolV2, ""), + newSUDPBridgeRW(t, &out, wire.ProtocolV1, ""), &count, nil, ) @@ -127,6 +166,22 @@ func TestSUDPBridgeRejectsUnknownVisitorMessage(t *testing.T) { require.Empty(t, out.Bytes()) } +func TestSUDPBridgeRejectsMismatchedPacketCodecOnStream(t *testing.T) { + var in, out bytes.Buffer + writeSUDPBridgeMsg(t, &in, wire.ProtocolV2, "", newSUDPBridgeUDPPacket("json-on-binary-stream")) + + var count int64 + err := bridgeSUDPProxyToVisitor( + newSUDPBridgeRW(t, &in, wire.ProtocolV2, wire.UDPPacketCodecBinary), + newSUDPBridgeRW(t, &out, wire.ProtocolV2, ""), + &count, + nil, + ) + require.ErrorContains(t, err, "received JSON UDP packet after binary codec negotiation") + require.Zero(t, count) + require.Empty(t, out.Bytes()) +} + func TestSUDPBridgeDetectsMixedWireProtocol(t *testing.T) { require.False(t, isMixedWireProtocol("", wire.ProtocolV1)) require.False(t, isMixedWireProtocol(wire.ProtocolV2, wire.ProtocolV2)) @@ -134,8 +189,170 @@ func TestSUDPBridgeDetectsMixedWireProtocol(t *testing.T) { require.True(t, isMixedWireProtocol(wire.ProtocolV2, wire.ProtocolV1)) } -func writeSUDPBridgeMsg(t *testing.T, buf *bytes.Buffer, wireProtocol string, m msg.Message) { - t.Helper() - - require.NoError(t, msg.NewReadWriter(buf, wireProtocol).WriteMsg(m)) +func TestSUDPBridgeDetectsMixedPacketEncoding(t *testing.T) { + for _, tc := range []struct { + name string + leftWire string + leftCodec string + rightWire string + rightCodec string + mixed bool + }{ + {name: "legacy v1 aliases explicit v1", leftWire: "", rightWire: wire.ProtocolV1}, + {name: "v2 json matches v2 json", leftWire: wire.ProtocolV2, rightWire: wire.ProtocolV2}, + { + name: "v2 binary matches v2 binary", + leftWire: wire.ProtocolV2, + leftCodec: wire.UDPPacketCodecBinary, + rightWire: wire.ProtocolV2, + rightCodec: wire.UDPPacketCodecBinary, + }, + {name: "v1 json differs from v2 json", leftWire: wire.ProtocolV1, rightWire: wire.ProtocolV2, mixed: true}, + { + name: "v2 json differs from v2 binary", + leftWire: wire.ProtocolV2, + rightWire: wire.ProtocolV2, + rightCodec: wire.UDPPacketCodecBinary, + mixed: true, + }, + { + name: "v2 binary differs from v1 json", + leftWire: wire.ProtocolV2, + leftCodec: wire.UDPPacketCodecBinary, + rightWire: wire.ProtocolV1, + mixed: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + mixed, err := isMixedSUDPPacketEncoding(tc.leftWire, tc.leftCodec, tc.rightWire, tc.rightCodec) + require.NoError(t, err) + require.Equal(t, tc.mixed, mixed) + }) + } +} + +func TestSUDPBridgeRejectsInvalidEncodingMetadata(t *testing.T) { + for _, tc := range []struct { + name string + leftWire string + leftCodec string + rightWire string + rightCodec string + wantErr string + }{ + { + name: "left v1 binary", + leftWire: wire.ProtocolV1, + leftCodec: wire.UDPPacketCodecBinary, + rightWire: wire.ProtocolV1, + wantErr: "invalid left SUDP packet encoding", + }, + { + name: "right unknown v2 codec", + leftWire: wire.ProtocolV2, + rightWire: wire.ProtocolV2, + rightCodec: "snappy", + wantErr: "invalid right SUDP packet encoding", + }, + {name: "left unknown wire", leftWire: "v3", rightWire: wire.ProtocolV2, wantErr: "unsupported wire protocol"}, + } { + t.Run(tc.name, func(t *testing.T) { + mixed, err := isMixedSUDPPacketEncoding(tc.leftWire, tc.leftCodec, tc.rightWire, tc.rightCodec) + require.False(t, mixed) + require.ErrorContains(t, err, tc.wantErr) + }) + } +} + +func TestSUDPJoinUsesRawPathForSameEncodingState(t *testing.T) { + proxyClient, proxyServer := net.Pipe() + visitorClient, visitorServer := net.Pipe() + t.Cleanup(func() { + _ = proxyClient.Close() + _ = proxyServer.Close() + _ = visitorClient.Close() + _ = visitorServer.Close() + }) + deadline := time.Now().Add(3 * time.Second) + require.NoError(t, proxyClient.SetDeadline(deadline)) + require.NoError(t, proxyServer.SetDeadline(deadline)) + require.NoError(t, visitorClient.SetDeadline(deadline)) + require.NoError(t, visitorServer.SetDeadline(deadline)) + + pxy := &BaseProxy{ + configurer: &v1.SUDPProxyConfig{}, + wireProtocol: wire.ProtocolV2, + udpPacketCodec: wire.UDPPacketCodecBinary, + } + visitorConn := &metadataConn{Conn: visitorServer, wireProtocol: wire.ProtocolV2, udpPacketCodec: wire.UDPPacketCodecBinary} + joinDone := make(chan []error, 1) + go func() { + _, _, errs := pxy.joinUserConnection(proxyServer, visitorConn, string(v1.ProxyTypeSUDP), xlog.New()) + joinDone <- errs + }() + + raw := []byte{0, 16, 0, 0, 0, 4, 0xde, 0xad, 0xbe, 0xef} + _, err := proxyClient.Write(raw) + require.NoError(t, err) + got := make([]byte, len(raw)) + _, err = io.ReadFull(visitorClient, got) + require.NoError(t, err) + require.Equal(t, raw, got) + + _ = proxyClient.Close() + _ = visitorClient.Close() + <-joinDone +} + +func newSUDPBridgeRW(t *testing.T, buf *bytes.Buffer, wireProtocol, udpPacketCodec string) msg.ReadWriter { + t.Helper() + rw, err := msg.NewUDPPacketReadWriter(buf, wireProtocol, udpPacketCodec) + require.NoError(t, err) + return rw +} + +func writeSUDPBridgeMsg(t *testing.T, buf *bytes.Buffer, wireProtocol, udpPacketCodec string, m msg.Message) { + t.Helper() + require.NoError(t, newSUDPBridgeRW(t, buf, wireProtocol, udpPacketCodec).WriteMsg(m)) +} + +func newSUDPBridgeUDPPacket(content string) *msg.UDPPacket { + return &msg.UDPPacket{ + Content: []byte(content), + RemoteAddr: &net.UDPAddr{IP: net.ParseIP("192.0.2.1"), Port: 12345}, + } +} + +func requireV2UDPPacketFrame(t *testing.T, buf *bytes.Buffer, wantType uint16, want *msg.UDPPacket) { + t.Helper() + frame, err := wire.NewConn(buf).ReadFrame() + require.NoError(t, err) + require.Equal(t, wire.FrameTypeMessage, frame.Type) + require.GreaterOrEqual(t, len(frame.Payload), 2) + require.Equal(t, wantType, binary.BigEndian.Uint16(frame.Payload[:2])) + var got *msg.UDPPacket + if wantType == msg.V2TypeUDPPacketBinary { + got, err = msg.DecodeUDPPacketBinary(frame.Payload[2:]) + } else { + var decoded msg.UDPPacket + err = msg.DecodeV2MessageFrameInto(frame, &decoded) + got = &decoded + } + require.NoError(t, err) + require.Equal(t, want.Content, got.Content) + require.Equal(t, want.RemoteAddr.String(), got.RemoteAddr.String()) +} + +type metadataConn struct { + net.Conn + wireProtocol string + udpPacketCodec string +} + +func (c *metadataConn) WireProtocol() string { + return c.wireProtocol +} + +func (c *metadataConn) UDPPacketCodec() string { + return c.udpPacketCodec } diff --git a/server/service.go b/server/service.go index e24acf66..7aba085e 100644 --- a/server/service.go +++ b/server/service.go @@ -885,14 +885,22 @@ func (svr *Service) RegisterWorkConn( } func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVisitorConn, wireProtocol string) error { - admit := func(visitorUser string) error { + admit := func(visitorUser, visitorWireProtocol, visitorUDPPacketCodec string) error { + if visitorWireProtocol == "" { + visitorWireProtocol = wireProtocol + } return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey, - newMsg.UseEncryption, newMsg.UseCompression, visitorUser, wireProtocol) + newMsg.UseEncryption, newMsg.UseCompression, visitorUser, visitorWireProtocol, visitorUDPPacketCodec) } // TODO(deprecation): Compatible with old versions, can be without runID, user is empty. In later versions, it will be mandatory to include runID. // If runID is required, it is not compatible with versions prior to v0.50.0. if newMsg.RunID != "" { - admitted, err := svr.ctlManager.admitVisitorByRunID(newMsg.RunID, admit) + admitted, err := svr.ctlManager.admitVisitorByRunID(newMsg.RunID, func(visitorUser, controlWireProtocol, controlUDPPacketCodec string) error { + if wireProtocol != controlWireProtocol { + return fmt.Errorf("visitor connection wire protocol mismatch: got %s want %s", wireProtocol, controlWireProtocol) + } + return admit(visitorUser, controlWireProtocol, controlUDPPacketCodec) + }) if err != nil { return err } @@ -901,5 +909,5 @@ func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVis } return nil } - return admit("") + return admit("", wireProtocol, "") } diff --git a/server/service_test.go b/server/service_test.go index cc7645f8..e35f11b9 100644 --- a/server/service_test.go +++ b/server/service_test.go @@ -428,20 +428,24 @@ func TestServiceVisitorAdmissionSerializesReplacement(t *testing.T) { } t.Cleanup(resume) type admissionResult struct { - admitted bool - user string - err error + admitted bool + user string + wireProtocol string + udpPacketCodec string + err error } admissionDone := make(chan admissionResult, 1) go func() { - var admittedUser string - admitted, admitErr := svr.ctlManager.admitVisitorByRunID("shared-run", func(user string) error { - admittedUser = user + result := admissionResult{} + result.admitted, result.err = svr.ctlManager.admitVisitorByRunID("shared-run", func(user, wireProtocol, udpPacketCodec string) error { + result.user = user + result.wireProtocol = wireProtocol + result.udpPacketCodec = udpPacketCodec close(admissionEntered) <-resumeAdmission return nil }) - admissionDone <- admissionResult{admitted: admitted, user: admittedUser, err: admitErr} + admissionDone <- result }() waitForSignal(t, admissionEntered, "visitor admission callback") runMu := currentRunGateForTest(svr.ctlManager, "shared-run") @@ -477,6 +481,8 @@ func TestServiceVisitorAdmissionSerializesReplacement(t *testing.T) { require.NoError(t, admission.err) require.True(t, admission.admitted) require.Equal(t, "old-user", admission.user) + require.Equal(t, wire.ProtocolV1, admission.wireProtocol) + require.Empty(t, admission.udpPacketCodec) replacement := waitForResult(t, replacementDone, "replacement") require.NoError(t, replacement.err) ctlB := replacement.ctl @@ -762,6 +768,81 @@ func TestServiceVisitorRoutingExcludesPendingUser(t *testing.T) { waitForControlDone(t, ctl) } +func TestServiceVisitorRoutingCarriesControlPacketCodec(t *testing.T) { + svr := newControlTestService(t) + listener, err := svr.rc.VisitorManager.Listen("visitor", "secret", []string{"visitor-user"}) + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + + controlConn := newDeadlineReadConn() + controlMsgConn := msg.NewConn(controlConn, msg.NewV2ReadWriter(controlConn)) + ctl, err := svr.RegisterControl(controlMsgConn, &msg.Login{ + RunID: "visitor-binary-run", + User: "visitor-user", + ClientID: "visitor-client", + ClientSpec: msg.ClientSpec{ + AlwaysAuthPass: true, + }, + }, true, wire.ProtocolV2, wire.UDPPacketCodecBinary) + require.NoError(t, err) + + timestamp := time.Now().Unix() + visitorMsg := &msg.NewVisitorConn{ + RunID: "visitor-binary-run", + ProxyName: "visitor", + Timestamp: timestamp, + SignKey: util.GetAuthKey("secret", timestamp), + } + require.NoError(t, svr.completeControlLogin(ctl, func() error { return nil })) + waitForSignal(t, controlConn.readStarted, "binary visitor control reader to start") + + runningConn := newCountingCloseConn() + require.NoError(t, svr.RegisterVisitorConn(runningConn, visitorMsg, wire.ProtocolV2)) + accepted, err := listener.Accept() + require.NoError(t, err) + metadata, ok := accepted.(interface { + WireProtocol() string + UDPPacketCodec() string + }) + require.True(t, ok) + require.Equal(t, wire.ProtocolV2, metadata.WireProtocol()) + require.Equal(t, wire.UDPPacketCodecBinary, metadata.UDPPacketCodec()) + require.NoError(t, accepted.Close()) + require.Equal(t, int64(1), runningConn.closeCount.Load()) + + mismatchConn := newCountingCloseConn() + err = svr.RegisterVisitorConn(mismatchConn, visitorMsg, wire.ProtocolV1) + require.ErrorContains(t, err, "visitor connection wire protocol mismatch") + require.NoError(t, mismatchConn.Close()) + require.Equal(t, int64(1), mismatchConn.closeCount.Load()) + + require.NoError(t, ctl.Close()) + waitForControlDone(t, ctl) +} + +func TestServiceVisitorRoutingLegacyFallsBackToJSONPacketCodec(t *testing.T) { + svr := newControlTestService(t) + listener, err := svr.rc.VisitorManager.Listen("visitor", "secret", []string{""}) + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + + timestamp := time.Now().Unix() + visitorMsg := &msg.NewVisitorConn{ + ProxyName: "visitor", + Timestamp: timestamp, + SignKey: util.GetAuthKey("secret", timestamp), + } + visitorConn := newCountingCloseConn() + require.NoError(t, svr.RegisterVisitorConn(visitorConn, visitorMsg, wire.ProtocolV2)) + accepted, err := listener.Accept() + require.NoError(t, err) + metadata, ok := accepted.(interface{ UDPPacketCodec() string }) + require.True(t, ok) + require.Empty(t, metadata.UDPPacketCodec()) + require.NoError(t, accepted.Close()) + require.Equal(t, int64(1), visitorConn.closeCount.Load()) +} + func newControlTestService(t *testing.T) *Service { t.Helper() cfg := &v1.ServerConfig{} diff --git a/server/visitor/visitor.go b/server/visitor/visitor.go index 11a44367..7776aad8 100644 --- a/server/visitor/visitor.go +++ b/server/visitor/visitor.go @@ -65,8 +65,12 @@ func (vm *Manager) Listen(name string, sk string, allowUsers []string) (*netpkg. func (vm *Manager) NewConn(name string, conn net.Conn, timestamp int64, signKey string, useEncryption bool, useCompression bool, visitorUser string, - wireProtocol string, + wireProtocol string, udpPacketCodecs ...string, ) (err error) { + udpPacketCodec := "" + if len(udpPacketCodecs) > 0 { + udpPacketCodec = udpPacketCodecs[0] + } vm.mu.RLock() defer vm.mu.RUnlock() @@ -93,8 +97,9 @@ func (vm *Manager) NewConn(name string, conn net.Conn, timestamp int64, signKey } visitorConn := netpkg.WrapReadWriteCloserToConn(rwc, conn) err = l.l.PutConn(&wireProtocolConn{ - Conn: visitorConn, - wireProtocol: wireProtocol, + Conn: visitorConn, + wireProtocol: wireProtocol, + udpPacketCodec: udpPacketCodec, }) } else { err = fmt.Errorf("custom listener for [%s] doesn't exist", name) @@ -105,13 +110,18 @@ func (vm *Manager) NewConn(name string, conn net.Conn, timestamp int64, signKey type wireProtocolConn struct { net.Conn - wireProtocol string + wireProtocol string + udpPacketCodec string } func (c *wireProtocolConn) WireProtocol() string { return c.wireProtocol } +func (c *wireProtocolConn) UDPPacketCodec() string { + return c.udpPacketCodec +} + func (vm *Manager) CloseListener(name string) { vm.mu.Lock() defer vm.mu.Unlock() diff --git a/server/visitor/visitor_test.go b/server/visitor/visitor_test.go index 67b907fd..04c50ff0 100644 --- a/server/visitor/visitor_test.go +++ b/server/visitor/visitor_test.go @@ -25,7 +25,7 @@ import ( "github.com/fatedier/frp/pkg/util/util" ) -func TestManagerNewConnCarriesWireProtocol(t *testing.T) { +func TestManagerNewConnCarriesWireProtocolAndUDPPacketCodec(t *testing.T) { vm := NewManager() listener, err := vm.Listen("sudp", "secret", []string{"*"}) require.NoError(t, err) @@ -47,6 +47,7 @@ func TestManagerNewConnCarriesWireProtocol(t *testing.T) { false, "user", wire.ProtocolV2, + wire.UDPPacketCodecBinary, ) }() @@ -54,8 +55,12 @@ func TestManagerNewConnCarriesWireProtocol(t *testing.T) { require.NoError(t, err) defer acceptedConn.Close() - getter, ok := acceptedConn.(interface{ WireProtocol() string }) + metadata, ok := acceptedConn.(interface { + WireProtocol() string + UDPPacketCodec() string + }) require.True(t, ok) - require.Equal(t, wire.ProtocolV2, getter.WireProtocol()) + require.Equal(t, wire.ProtocolV2, metadata.WireProtocol()) + require.Equal(t, wire.UDPPacketCodecBinary, metadata.UDPPacketCodec()) require.NoError(t, <-errCh) } diff --git a/test/e2e/v1/basic/wire.go b/test/e2e/v1/basic/wire.go index cb61558f..bf475287 100644 --- a/test/e2e/v1/basic/wire.go +++ b/test/e2e/v1/basic/wire.go @@ -110,12 +110,12 @@ var _ = ginkgo.Describe("[Feature: WireProtocol]", func() { name: "default sudp visitor", }, { - name: "v2 sudp visitor", + name: "v2 binary raw sudp visitor", proxyWireConfig: `transport.wireProtocol = "v2"`, visitorWireConfig: `transport.wireProtocol = "v2"`, }, { - name: "mixed sudp proxy v1 visitor v2", + name: "v1 JSON proxy -> v2 Binary visitor transcode", proxyWireConfig: `transport.wireProtocol = "v1"`, visitorWireConfig: `transport.wireProtocol = "v2"`, extraProxyConfig: ` @@ -128,7 +128,7 @@ var _ = ginkgo.Describe("[Feature: WireProtocol]", func() { `, }, { - name: "mixed sudp proxy v2 visitor v1", + name: "v2 Binary proxy -> v1 JSON visitor transcode", proxyWireConfig: `transport.wireProtocol = "v2"`, visitorWireConfig: `transport.wireProtocol = "v1"`, },