Compare commits

..

2 Commits

Author SHA1 Message Date
copilot-swe-agent[bot]
7993225be1 Initial plan 2026-03-05 17:33:24 +00:00
fatedier
17b27d8d96 pkg/msg: change UDPPacket.Content from string to []byte to avoid redundant base64 encode/decode 2026-03-06 01:25:52 +08:00
69 changed files with 791 additions and 617 deletions

View File

@@ -18,7 +18,6 @@ linters:
- lll - lll
- makezero - makezero
- misspell - misspell
- modernize
- prealloc - prealloc
- predeclared - predeclared
- revive - revive
@@ -48,9 +47,6 @@ linters:
ignore-rules: ignore-rules:
- cancelled - cancelled
- marshalled - marshalled
modernize:
disable:
- omitzero
unparam: unparam:
check-exported: false check-exported: false
exclusions: exclusions:

View File

@@ -16,7 +16,6 @@ package proxy
import ( import (
"context" "context"
"fmt"
"io" "io"
"net" "net"
"reflect" "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) { func (pxy *BaseProxy) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) {
pxy.inWorkConnCallback = cb pxy.inWorkConnCallback = cb
} }
@@ -167,14 +139,30 @@ func (pxy *BaseProxy) InWorkConn(conn net.Conn, m *msg.StartWorkConn) {
func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWorkConn, encKey []byte) { func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWorkConn, encKey []byte) {
xl := pxy.xl xl := pxy.xl
baseCfg := pxy.baseCfg 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", xl.Tracef("handle tcp work connection, useEncryption: %t, useCompression: %t",
baseCfg.Transport.UseEncryption, baseCfg.Transport.UseCompression) 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 { if err != nil {
xl.Errorf("wrap work connection: %v", err) workConn.Close()
return 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 // check if we need to send proxy protocol info
@@ -190,6 +178,7 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
} }
if baseCfg.Transport.ProxyProtocolVersion != "" && m.SrcAddr != "" && m.SrcPort != 0 { 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) header := netpkg.BuildProxyProtocolHeaderStruct(connInfo.SrcAddr, connInfo.DstAddr, baseCfg.Transport.ProxyProtocolVersion)
connInfo.ProxyProtocolHeader = header connInfo.ProxyProtocolHeader = header
} }
@@ -198,18 +187,12 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
if pxy.proxyPlugin != nil { if pxy.proxyPlugin != nil {
// if plugin is set, let plugin handle connection first // 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()) xl.Debugf("handle by plugin: %s", pxy.proxyPlugin.Name())
pxy.proxyPlugin.Handle(pxy.ctx, &connInfo) pxy.proxyPlugin.Handle(pxy.ctx, &connInfo)
xl.Debugf("handle by plugin finished") xl.Debugf("handle by plugin finished")
return return
} }
if recycleFn != nil {
defer recycleFn()
}
localConn, err := libnet.Dial( localConn, err := libnet.Dial(
net.JoinHostPort(baseCfg.LocalIP, strconv.Itoa(baseCfg.LocalPort)), net.JoinHostPort(baseCfg.LocalIP, strconv.Itoa(baseCfg.LocalPort)),
libnet.WithTimeout(10*time.Second), libnet.WithTimeout(10*time.Second),
@@ -226,7 +209,6 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
if connInfo.ProxyProtocolHeader != nil { if connInfo.ProxyProtocolHeader != nil {
if _, err := connInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil { if _, err := connInfo.ProxyProtocolHeader.WriteTo(localConn); err != nil {
workConn.Close() workConn.Close()
localConn.Close()
xl.Errorf("write proxy protocol header to local conn error: %v", err) xl.Errorf("write proxy protocol header to local conn error: %v", err)
return return
} }
@@ -237,4 +219,7 @@ func (pxy *BaseProxy) HandleTCPWorkConnection(workConn net.Conn, m *msg.StartWor
if len(errs) > 0 { if len(errs) > 0 {
xl.Tracef("join connections errors: %v", errs) xl.Tracef("join connections errors: %v", errs)
} }
if compressionResourceRecycleFn != nil {
compressionResourceRecycleFn()
}
} }

View File

@@ -17,6 +17,7 @@
package proxy package proxy
import ( import (
"io"
"net" "net"
"reflect" "reflect"
"strconv" "strconv"
@@ -24,15 +25,17 @@ import (
"time" "time"
"github.com/fatedier/golib/errors" "github.com/fatedier/golib/errors"
libio "github.com/fatedier/golib/io"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/msg" "github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/proto/udp" "github.com/fatedier/frp/pkg/proto/udp"
"github.com/fatedier/frp/pkg/util/limit"
netpkg "github.com/fatedier/frp/pkg/util/net" netpkg "github.com/fatedier/frp/pkg/util/net"
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.SUDPProxyConfig](), NewSUDPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.SUDPProxyConfig{}), NewSUDPProxy)
} }
type SUDPProxy struct { type SUDPProxy struct {
@@ -80,13 +83,27 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
xl := pxy.xl xl := pxy.xl
xl.Infof("incoming a new work connection for sudp proxy, %s", conn.RemoteAddr().String()) 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
if err != nil { var err error
xl.Errorf("wrap work connection: %v", err) if pxy.limiter != nil {
return 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 {
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) readCh := make(chan *msg.UDPPacket, 1024)
sendCh := make(chan msg.Message, 1024) sendCh := make(chan msg.Message, 1024)
isClose := false isClose := false

View File

@@ -17,21 +17,24 @@
package proxy package proxy
import ( import (
"io"
"net" "net"
"reflect" "reflect"
"strconv" "strconv"
"time" "time"
"github.com/fatedier/golib/errors" "github.com/fatedier/golib/errors"
libio "github.com/fatedier/golib/io"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/msg" "github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/proto/udp" "github.com/fatedier/frp/pkg/proto/udp"
"github.com/fatedier/frp/pkg/util/limit"
netpkg "github.com/fatedier/frp/pkg/util/net" netpkg "github.com/fatedier/frp/pkg/util/net"
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.UDPProxyConfig](), NewUDPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.UDPProxyConfig{}), NewUDPProxy)
} }
type UDPProxy struct { type UDPProxy struct {
@@ -91,14 +94,28 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
// close resources related with old workConn // close resources related with old workConn
pxy.Close() pxy.Close()
remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey) var rwc io.ReadWriteCloser = conn
if err != nil { var err error
xl.Errorf("wrap work connection: %v", err) if pxy.limiter != nil {
return 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 {
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.mu.Lock()
pxy.workConn = netpkg.WrapReadWriteCloserToConn(remote, conn) pxy.workConn = conn
pxy.readCh = make(chan *msg.UDPPacket, 1024) pxy.readCh = make(chan *msg.UDPPacket, 1024)
pxy.sendCh = make(chan msg.Message, 1024) pxy.sendCh = make(chan msg.Message, 1024)
pxy.closed = false pxy.closed = false

View File

@@ -34,7 +34,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.XTCPProxyConfig](), NewXTCPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.XTCPProxyConfig{}), NewXTCPProxy)
} }
type XTCPProxy struct { type XTCPProxy struct {

View File

@@ -15,12 +15,18 @@
package visitor package visitor
import ( import (
"fmt"
"io"
"net" "net"
"strconv" "strconv"
"time"
libio "github.com/fatedier/golib/io" libio "github.com/fatedier/golib/io"
v1 "github.com/fatedier/frp/pkg/config/v1" 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/util/util"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
) )
@@ -36,10 +42,10 @@ func (sv *STCPVisitor) Run() (err error) {
if err != nil { if err != nil {
return 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 { if sv.plugin != nil {
sv.plugin.Start() sv.plugin.Start()
@@ -51,10 +57,35 @@ func (sv *STCPVisitor) Close() {
sv.BaseVisitor.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) { func (sv *STCPVisitor) handleConn(userConn net.Conn) {
xl := xlog.FromContextSafe(sv.ctx) xl := xlog.FromContextSafe(sv.ctx)
var tunnelErr error var tunnelErr error
defer func() { defer func() {
// If there was an error and connection supports CloseWithError, use it
if tunnelErr != nil { if tunnelErr != nil {
if eConn, ok := userConn.(interface{ CloseWithError(error) error }); ok { if eConn, ok := userConn.(interface{ CloseWithError(error) error }); ok {
_ = eConn.CloseWithError(tunnelErr) _ = eConn.CloseWithError(tunnelErr)
@@ -65,21 +96,62 @@ func (sv *STCPVisitor) handleConn(userConn net.Conn) {
}() }()
xl.Debugf("get a new stcp user connection") xl.Debugf("get a new stcp user connection")
visitorConn, err := sv.dialRawVisitorConn(sv.cfg.GetBaseConfig()) visitorConn, err := sv.helper.ConnectServer()
if err != nil { if err != nil {
xl.Warnf("dialRawVisitorConn error: %v", err)
tunnelErr = err tunnelErr = err
return return
} }
defer visitorConn.Close() defer visitorConn.Close()
remote, recycleFn, err := wrapVisitorConn(visitorConn, sv.cfg.GetBaseConfig()) now := time.Now().Unix()
targetProxyName := naming.BuildTargetServerProxyName(sv.clientCfg.User, sv.cfg.ServerUser, sv.cfg.ServerName)
newVisitorConnMsg := &msg.NewVisitorConn{
RunID: sv.helper.RunID(),
ProxyName: targetProxyName,
SignKey: util.GetAuthKey(sv.cfg.SecretKey, now),
Timestamp: now,
UseEncryption: sv.cfg.Transport.UseEncryption,
UseCompression: sv.cfg.Transport.UseCompression,
}
err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
if err != nil { if err != nil {
xl.Warnf("wrapVisitorConn error: %v", err) xl.Warnf("send newVisitorConnMsg to server error: %v", err)
tunnelErr = err tunnelErr = err
return return
} }
defer recycleFn()
var newVisitorConnRespMsg msg.NewVisitorConnResp
_ = visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
if err != nil {
xl.Warnf("get newVisitorConnRespMsg error: %v", err)
tunnelErr = err
return
}
_ = visitorConn.SetReadDeadline(time.Time{})
if newVisitorConnRespMsg.Error != "" {
xl.Warnf("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
tunnelErr = fmt.Errorf("%s", newVisitorConnRespMsg.Error)
return
}
var remote io.ReadWriteCloser
remote = visitorConn
if sv.cfg.Transport.UseEncryption {
remote, err = libio.WithEncryption(remote, []byte(sv.cfg.SecretKey))
if err != nil {
xl.Errorf("create encryption stream error: %v", err)
tunnelErr = err
return
}
}
if sv.cfg.Transport.UseCompression {
var recycleFn func()
remote, recycleFn = libio.WithCompressionFromPool(remote)
defer recycleFn()
}
libio.Join(userConn, remote) libio.Join(userConn, remote)
} }

View File

@@ -16,17 +16,21 @@ package visitor
import ( import (
"fmt" "fmt"
"io"
"net" "net"
"strconv" "strconv"
"sync" "sync"
"time" "time"
"github.com/fatedier/golib/errors" "github.com/fatedier/golib/errors"
libio "github.com/fatedier/golib/io"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/msg" "github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/naming"
"github.com/fatedier/frp/pkg/proto/udp" "github.com/fatedier/frp/pkg/proto/udp"
netpkg "github.com/fatedier/frp/pkg/util/net" netpkg "github.com/fatedier/frp/pkg/util/net"
"github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
) )
@@ -72,7 +76,6 @@ func (sv *SUDPVisitor) dispatcher() {
var ( var (
visitorConn net.Conn visitorConn net.Conn
recycleFn func()
err error err error
firstPacket *msg.UDPPacket firstPacket *msg.UDPPacket
@@ -90,17 +93,14 @@ func (sv *SUDPVisitor) dispatcher() {
return return
} }
visitorConn, recycleFn, err = sv.getNewVisitorConn() visitorConn, err = sv.getNewVisitorConn()
if err != nil { if err != nil {
xl.Warnf("newVisitorConn to frps error: %v, try to reconnect", err) xl.Warnf("newVisitorConn to frps error: %v, try to reconnect", err)
continue continue
} }
// visitorConn always be closed when worker done. // visitorConn always be closed when worker done.
func() { sv.worker(visitorConn, firstPacket)
defer recycleFn()
sv.worker(visitorConn, firstPacket)
}()
select { select {
case <-sv.checkCloseCh: case <-sv.checkCloseCh:
@@ -198,17 +198,53 @@ func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
xl.Infof("sudp worker is closed") xl.Infof("sudp worker is closed")
} }
func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, func(), error) { func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, error) {
rawConn, err := sv.dialRawVisitorConn(sv.cfg.GetBaseConfig()) xl := xlog.FromContextSafe(sv.ctx)
visitorConn, err := sv.helper.ConnectServer()
if err != nil { if err != nil {
return nil, func() {}, err return nil, fmt.Errorf("frpc connect frps error: %v", err)
} }
rwc, recycleFn, err := wrapVisitorConn(rawConn, sv.cfg.GetBaseConfig())
now := time.Now().Unix()
targetProxyName := naming.BuildTargetServerProxyName(sv.clientCfg.User, sv.cfg.ServerUser, sv.cfg.ServerName)
newVisitorConnMsg := &msg.NewVisitorConn{
RunID: sv.helper.RunID(),
ProxyName: targetProxyName,
SignKey: util.GetAuthKey(sv.cfg.SecretKey, now),
Timestamp: now,
UseEncryption: sv.cfg.Transport.UseEncryption,
UseCompression: sv.cfg.Transport.UseCompression,
}
err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
if err != nil { if err != nil {
rawConn.Close() return nil, fmt.Errorf("frpc send newVisitorConnMsg to frps error: %v", err)
return nil, func() {}, err
} }
return netpkg.WrapReadWriteCloserToConn(rwc, rawConn), recycleFn, nil
var newVisitorConnRespMsg msg.NewVisitorConnResp
_ = visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
if err != nil {
return nil, fmt.Errorf("frpc read newVisitorConnRespMsg error: %v", err)
}
_ = visitorConn.SetReadDeadline(time.Time{})
if newVisitorConnRespMsg.Error != "" {
return nil, fmt.Errorf("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
}
var remote io.ReadWriteCloser
remote = visitorConn
if sv.cfg.Transport.UseEncryption {
remote, err = libio.WithEncryption(remote, []byte(sv.cfg.SecretKey))
if err != nil {
xl.Errorf("create encryption stream error: %v", err)
return nil, err
}
}
if sv.cfg.Transport.UseCompression {
remote = libio.WithCompression(remote)
}
return netpkg.WrapReadWriteCloserToConn(remote, visitorConn), nil
} }
func (sv *SUDPVisitor) Close() { func (sv *SUDPVisitor) Close() {

View File

@@ -16,21 +16,13 @@ package visitor
import ( import (
"context" "context"
"fmt"
"io"
"net" "net"
"sync" "sync"
"time"
libio "github.com/fatedier/golib/io"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
"github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/naming"
plugin "github.com/fatedier/frp/pkg/plugin/visitor" plugin "github.com/fatedier/frp/pkg/plugin/visitor"
"github.com/fatedier/frp/pkg/transport" "github.com/fatedier/frp/pkg/transport"
netpkg "github.com/fatedier/frp/pkg/util/net" netpkg "github.com/fatedier/frp/pkg/util/net"
"github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
"github.com/fatedier/frp/pkg/vnet" "github.com/fatedier/frp/pkg/vnet"
) )
@@ -127,18 +119,6 @@ func (v *BaseVisitor) AcceptConn(conn net.Conn) error {
return v.internalLn.PutConn(conn) 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() { func (v *BaseVisitor) Close() {
if v.l != nil { if v.l != nil {
v.l.Close() v.l.Close()
@@ -150,57 +130,3 @@ func (v *BaseVisitor) Close() {
v.plugin.Close() v.plugin.Close()
} }
} }
func (v *BaseVisitor) dialRawVisitorConn(cfg *v1.VisitorBaseConfig) (net.Conn, error) {
visitorConn, err := v.helper.ConnectServer()
if err != nil {
return nil, fmt.Errorf("connect to server error: %v", err)
}
now := time.Now().Unix()
targetProxyName := naming.BuildTargetServerProxyName(v.clientCfg.User, cfg.ServerUser, cfg.ServerName)
newVisitorConnMsg := &msg.NewVisitorConn{
RunID: v.helper.RunID(),
ProxyName: targetProxyName,
SignKey: util.GetAuthKey(cfg.SecretKey, now),
Timestamp: now,
UseEncryption: cfg.Transport.UseEncryption,
UseCompression: cfg.Transport.UseCompression,
}
err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
if err != nil {
visitorConn.Close()
return nil, fmt.Errorf("send newVisitorConnMsg to server error: %v", err)
}
var newVisitorConnRespMsg msg.NewVisitorConnResp
_ = visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
if err != nil {
visitorConn.Close()
return nil, fmt.Errorf("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)
}
return visitorConn, nil
}
func wrapVisitorConn(conn io.ReadWriteCloser, cfg *v1.VisitorBaseConfig) (io.ReadWriteCloser, func(), error) {
rwc := conn
if cfg.Transport.UseEncryption {
var err error
rwc, err = libio.WithEncryption(rwc, []byte(cfg.SecretKey))
if err != nil {
return nil, func() {}, fmt.Errorf("create encryption stream error: %v", err)
}
}
recycleFn := func() {}
if cfg.Transport.UseCompression {
rwc, recycleFn = libio.WithCompressionFromPool(rwc)
}
return rwc, recycleFn, nil
}

View File

@@ -65,10 +65,10 @@ func (sv *XTCPVisitor) Run() (err error) {
if err != nil { if err != nil {
return 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() go sv.processTunnelStartEvents()
if sv.cfg.KeepTunnelOpen { if sv.cfg.KeepTunnelOpen {
sv.retryLimiter = rate.NewLimiter(rate.Every(time.Hour/time.Duration(sv.cfg.MaxRetriesAnHour)), sv.cfg.MaxRetriesAnHour) 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() { func (sv *XTCPVisitor) processTunnelStartEvents() {
for { for {
select { select {
@@ -182,14 +206,20 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
return return
} }
muxConnRWCloser, recycleFn, err := wrapVisitorConn(tunnelConn, sv.cfg.GetBaseConfig()) var muxConnRWCloser io.ReadWriteCloser = tunnelConn
if err != nil { if sv.cfg.Transport.UseEncryption {
xl.Errorf("%v", err) muxConnRWCloser, err = libio.WithEncryption(muxConnRWCloser, []byte(sv.cfg.SecretKey))
tunnelConn.Close() if err != nil {
tunnelErr = err xl.Errorf("create encryption stream error: %v", err)
return tunnelErr = err
return
}
}
if sv.cfg.Transport.UseCompression {
var recycleFn func()
muxConnRWCloser, recycleFn = libio.WithCompressionFromPool(muxConnRWCloser)
defer recycleFn()
} }
defer recycleFn()
_, _, errs := libio.Join(userConn, muxConnRWCloser) _, _, errs := libio.Join(userConn, muxConnRWCloser)
xl.Debugf("join connections closed") xl.Debugf("join connections closed")
@@ -343,7 +373,6 @@ func (ks *KCPTunnelSession) Init(listenConn *net.UDPConn, raddr *net.UDPAddr) er
} }
remote, err := netpkg.NewKCPConnFromUDP(lConn, true, raddr.String()) remote, err := netpkg.NewKCPConnFromUDP(lConn, true, raddr.String())
if err != nil { if err != nil {
lConn.Close()
return fmt.Errorf("create kcp connection from udp connection error: %v", err) return fmt.Errorf("create kcp connection from udp connection error: %v", err)
} }

View File

@@ -47,7 +47,7 @@ var natholeDiscoveryCmd = &cobra.Command{
Use: "discover", Use: "discover",
Short: "Discover nathole information from stun server", Short: "Discover nathole information from stun server",
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
// ignore error here, because we can use command line parameters // ignore error here, because we can use command line pameters
cfg, _, _, _, err := config.LoadClientConfig(cfgFile, strictConfigMode) cfg, _, _, _, err := config.LoadClientConfig(cfgFile, strictConfigMode)
if err != nil { if err != nil {
cfg = &v1.ClientCommonConfig{} cfg = &v1.ClientCommonConfig{}

View File

@@ -23,7 +23,6 @@ import (
"net/url" "net/url"
"os" "os"
"slices" "slices"
"sync"
"github.com/coreos/go-oidc/v3/oidc" "github.com/coreos/go-oidc/v3/oidc"
"golang.org/x/oauth2" "golang.org/x/oauth2"
@@ -206,8 +205,7 @@ type OidcAuthConsumer struct {
additionalAuthScopes []v1.AuthScope additionalAuthScopes []v1.AuthScope
verifier TokenVerifier verifier TokenVerifier
mu sync.RWMutex subjectsFromLogin []string
subjectsFromLogin map[string]struct{}
} }
func NewTokenVerifier(cfg v1.AuthOIDCServerConfig) TokenVerifier { func NewTokenVerifier(cfg v1.AuthOIDCServerConfig) TokenVerifier {
@@ -228,7 +226,7 @@ func NewOidcAuthVerifier(additionalAuthScopes []v1.AuthScope, verifier TokenVeri
return &OidcAuthConsumer{ return &OidcAuthConsumer{
additionalAuthScopes: additionalAuthScopes, additionalAuthScopes: additionalAuthScopes,
verifier: verifier, 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 { if err != nil {
return fmt.Errorf("invalid OIDC token in login: %v", err) 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 return nil
} }
@@ -248,13 +246,11 @@ func (auth *OidcAuthConsumer) verifyPostLoginToken(privilegeKey string) (err err
if err != nil { if err != nil {
return fmt.Errorf("invalid OIDC token in ping: %v", err) 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. "+ return fmt.Errorf("received different OIDC subject in login and ping. "+
"original subjects: %s, "+
"new subject: %s", "new subject: %s",
token.Subject) auth.subjectsFromLogin, token.Subject)
} }
return nil return nil
} }

View File

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

View File

@@ -39,14 +39,14 @@ const (
// Proxy // Proxy
var ( var (
proxyConfTypeMap = map[ProxyType]reflect.Type{ proxyConfTypeMap = map[ProxyType]reflect.Type{
ProxyTypeTCP: reflect.TypeFor[TCPProxyConf](), ProxyTypeTCP: reflect.TypeOf(TCPProxyConf{}),
ProxyTypeUDP: reflect.TypeFor[UDPProxyConf](), ProxyTypeUDP: reflect.TypeOf(UDPProxyConf{}),
ProxyTypeTCPMUX: reflect.TypeFor[TCPMuxProxyConf](), ProxyTypeTCPMUX: reflect.TypeOf(TCPMuxProxyConf{}),
ProxyTypeHTTP: reflect.TypeFor[HTTPProxyConf](), ProxyTypeHTTP: reflect.TypeOf(HTTPProxyConf{}),
ProxyTypeHTTPS: reflect.TypeFor[HTTPSProxyConf](), ProxyTypeHTTPS: reflect.TypeOf(HTTPSProxyConf{}),
ProxyTypeSTCP: reflect.TypeFor[STCPProxyConf](), ProxyTypeSTCP: reflect.TypeOf(STCPProxyConf{}),
ProxyTypeXTCP: reflect.TypeFor[XTCPProxyConf](), ProxyTypeXTCP: reflect.TypeOf(XTCPProxyConf{}),
ProxyTypeSUDP: reflect.TypeFor[SUDPProxyConf](), ProxyTypeSUDP: reflect.TypeOf(SUDPProxyConf{}),
} }
) )

View File

@@ -22,8 +22,8 @@ func GetMapWithoutPrefix(set map[string]string, prefix string) map[string]string
m := make(map[string]string) m := make(map[string]string)
for key, value := range set { 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
} }
} }

View File

@@ -32,9 +32,9 @@ const (
// Visitor // Visitor
var ( var (
visitorConfTypeMap = map[VisitorType]reflect.Type{ visitorConfTypeMap = map[VisitorType]reflect.Type{
VisitorTypeSTCP: reflect.TypeFor[STCPVisitorConf](), VisitorTypeSTCP: reflect.TypeOf(STCPVisitorConf{}),
VisitorTypeXTCP: reflect.TypeFor[XTCPVisitorConf](), VisitorTypeXTCP: reflect.TypeOf(XTCPVisitorConf{}),
VisitorTypeSUDP: reflect.TypeFor[SUDPVisitorConf](), VisitorTypeSUDP: reflect.TypeOf(SUDPVisitorConf{}),
} }
) )

View File

@@ -15,11 +15,9 @@
package source package source
import ( import (
"cmp"
"errors" "errors"
"fmt" "fmt"
"maps" "sort"
"slices"
"sync" "sync"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
@@ -99,11 +97,21 @@ func (a *Aggregator) mapsToSortedSlices(
proxyMap map[string]v1.ProxyConfigurer, proxyMap map[string]v1.ProxyConfigurer,
visitorMap map[string]v1.VisitorConfigurer, visitorMap map[string]v1.VisitorConfigurer,
) ([]v1.ProxyConfigurer, []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 return proxies, visitors
} }

View File

@@ -196,27 +196,6 @@ func TestAggregator_VisitorMerge(t *testing.T) {
require.Len(visitors, 2) 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) { func TestAggregator_Load_ReturnsDefensiveCopies(t *testing.T) {
require := require.New(t) require := require.New(t)

View File

@@ -38,7 +38,7 @@ func parseNumberRangePair(firstRangeStr, secondRangeStr string) ([]NumberPair, e
return nil, fmt.Errorf("first and second range numbers are not in pairs") return nil, fmt.Errorf("first and second range numbers are not in pairs")
} }
pairs := make([]NumberPair, 0, len(firstRangeNumbers)) pairs := make([]NumberPair, 0, len(firstRangeNumbers))
for i := range firstRangeNumbers { for i := 0; i < len(firstRangeNumbers); i++ {
pairs = append(pairs, NumberPair{ pairs = append(pairs, NumberPair{
First: firstRangeNumbers[i], First: firstRangeNumbers[i],
Second: secondRangeNumbers[i], Second: secondRangeNumbers[i],

View File

@@ -70,18 +70,24 @@ func (q *BandwidthQuantity) UnmarshalString(s string) error {
f float64 f float64
err error err error
) )
if fstr, ok := strings.CutSuffix(s, "MB"); ok { switch {
case strings.HasSuffix(s, "MB"):
base = MB base = MB
fstr := strings.TrimSuffix(s, "MB")
f, err = strconv.ParseFloat(fstr, 64) f, err = strconv.ParseFloat(fstr, 64)
} else if fstr, ok := strings.CutSuffix(s, "KB"); ok { if err != nil {
return err
}
case strings.HasSuffix(s, "KB"):
base = KB base = KB
fstr := strings.TrimSuffix(s, "KB")
f, err = strconv.ParseFloat(fstr, 64) f, err = strconv.ParseFloat(fstr, 64)
} else { if err != nil {
return err
}
default:
return errors.New("unit not support") return errors.New("unit not support")
} }
if err != nil {
return err
}
q.s = s q.s = s
q.i = int64(f * float64(base)) q.i = int64(f * float64(base))
@@ -137,8 +143,8 @@ func (p PortsRangeSlice) String() string {
func NewPortsRangeSliceFromString(str string) ([]PortsRange, error) { func NewPortsRangeSliceFromString(str string) ([]PortsRange, error) {
str = strings.TrimSpace(str) str = strings.TrimSpace(str)
out := []PortsRange{} out := []PortsRange{}
numRanges := strings.SplitSeq(str, ",") numRanges := strings.Split(str, ",")
for numRangeStr := range numRanges { for _, numRangeStr := range numRanges {
// 1000-2000 or 2001 // 1000-2000 or 2001
numArray := strings.Split(numRangeStr, "-") numArray := strings.Split(numRangeStr, "-")
// length: only 1 or 2 is correct // length: only 1 or 2 is correct

View File

@@ -39,31 +39,6 @@ func TestBandwidthQuantity(t *testing.T) {
require.Equal(`{"b":"1KB","int":5}`, string(buf)) 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) { func TestPortsRangeSlice2String(t *testing.T) {
require := require.New(t) require := require.New(t)

View File

@@ -239,14 +239,14 @@ const (
) )
var proxyConfigTypeMap = map[ProxyType]reflect.Type{ var proxyConfigTypeMap = map[ProxyType]reflect.Type{
ProxyTypeTCP: reflect.TypeFor[TCPProxyConfig](), ProxyTypeTCP: reflect.TypeOf(TCPProxyConfig{}),
ProxyTypeUDP: reflect.TypeFor[UDPProxyConfig](), ProxyTypeUDP: reflect.TypeOf(UDPProxyConfig{}),
ProxyTypeHTTP: reflect.TypeFor[HTTPProxyConfig](), ProxyTypeHTTP: reflect.TypeOf(HTTPProxyConfig{}),
ProxyTypeHTTPS: reflect.TypeFor[HTTPSProxyConfig](), ProxyTypeHTTPS: reflect.TypeOf(HTTPSProxyConfig{}),
ProxyTypeTCPMUX: reflect.TypeFor[TCPMuxProxyConfig](), ProxyTypeTCPMUX: reflect.TypeOf(TCPMuxProxyConfig{}),
ProxyTypeSTCP: reflect.TypeFor[STCPProxyConfig](), ProxyTypeSTCP: reflect.TypeOf(STCPProxyConfig{}),
ProxyTypeXTCP: reflect.TypeFor[XTCPProxyConfig](), ProxyTypeXTCP: reflect.TypeOf(XTCPProxyConfig{}),
ProxyTypeSUDP: reflect.TypeFor[SUDPProxyConfig](), ProxyTypeSUDP: reflect.TypeOf(SUDPProxyConfig{}),
} }
func NewProxyConfigurerByType(proxyType ProxyType) ProxyConfigurer { func NewProxyConfigurerByType(proxyType ProxyType) ProxyConfigurer {

View File

@@ -37,16 +37,16 @@ const (
) )
var clientPluginOptionsTypeMap = map[string]reflect.Type{ var clientPluginOptionsTypeMap = map[string]reflect.Type{
PluginHTTP2HTTPS: reflect.TypeFor[HTTP2HTTPSPluginOptions](), PluginHTTP2HTTPS: reflect.TypeOf(HTTP2HTTPSPluginOptions{}),
PluginHTTPProxy: reflect.TypeFor[HTTPProxyPluginOptions](), PluginHTTPProxy: reflect.TypeOf(HTTPProxyPluginOptions{}),
PluginHTTPS2HTTP: reflect.TypeFor[HTTPS2HTTPPluginOptions](), PluginHTTPS2HTTP: reflect.TypeOf(HTTPS2HTTPPluginOptions{}),
PluginHTTPS2HTTPS: reflect.TypeFor[HTTPS2HTTPSPluginOptions](), PluginHTTPS2HTTPS: reflect.TypeOf(HTTPS2HTTPSPluginOptions{}),
PluginHTTP2HTTP: reflect.TypeFor[HTTP2HTTPPluginOptions](), PluginHTTP2HTTP: reflect.TypeOf(HTTP2HTTPPluginOptions{}),
PluginSocks5: reflect.TypeFor[Socks5PluginOptions](), PluginSocks5: reflect.TypeOf(Socks5PluginOptions{}),
PluginStaticFile: reflect.TypeFor[StaticFilePluginOptions](), PluginStaticFile: reflect.TypeOf(StaticFilePluginOptions{}),
PluginUnixDomainSocket: reflect.TypeFor[UnixDomainSocketPluginOptions](), PluginUnixDomainSocket: reflect.TypeOf(UnixDomainSocketPluginOptions{}),
PluginTLS2Raw: reflect.TypeFor[TLS2RawPluginOptions](), PluginTLS2Raw: reflect.TypeOf(TLS2RawPluginOptions{}),
PluginVirtualNet: reflect.TypeFor[VirtualNetPluginOptions](), PluginVirtualNet: reflect.TypeOf(VirtualNetPluginOptions{}),
} }
type ClientPluginOptions interface { type ClientPluginOptions interface {

View File

@@ -79,9 +79,9 @@ const (
) )
var visitorConfigTypeMap = map[VisitorType]reflect.Type{ var visitorConfigTypeMap = map[VisitorType]reflect.Type{
VisitorTypeSTCP: reflect.TypeFor[STCPVisitorConfig](), VisitorTypeSTCP: reflect.TypeOf(STCPVisitorConfig{}),
VisitorTypeXTCP: reflect.TypeFor[XTCPVisitorConfig](), VisitorTypeXTCP: reflect.TypeOf(XTCPVisitorConfig{}),
VisitorTypeSUDP: reflect.TypeFor[SUDPVisitorConfig](), VisitorTypeSUDP: reflect.TypeOf(SUDPVisitorConfig{}),
} }
type TypedVisitorConfig struct { type TypedVisitorConfig struct {

View File

@@ -25,7 +25,7 @@ const (
) )
var visitorPluginOptionsTypeMap = map[string]reflect.Type{ var visitorPluginOptionsTypeMap = map[string]reflect.Type{
VisitorPluginVirtualNet: reflect.TypeFor[VirtualNetVisitorPluginOptions](), VisitorPluginVirtualNet: reflect.TypeOf(VirtualNetVisitorPluginOptions{}),
} }
type VisitorPluginOptions interface { type VisitorPluginOptions interface {

View File

@@ -143,6 +143,7 @@ func (m *serverMetrics) OpenConnection(name string, _ string) {
proxyStats, ok := m.info.ProxyStatistics[name] proxyStats, ok := m.info.ProxyStatistics[name]
if ok { if ok {
proxyStats.CurConns.Inc(1) proxyStats.CurConns.Inc(1)
m.info.ProxyStatistics[name] = proxyStats
} }
} }
@@ -154,6 +155,7 @@ func (m *serverMetrics) CloseConnection(name string, _ string) {
proxyStats, ok := m.info.ProxyStatistics[name] proxyStats, ok := m.info.ProxyStatistics[name]
if ok { if ok {
proxyStats.CurConns.Dec(1) proxyStats.CurConns.Dec(1)
m.info.ProxyStatistics[name] = proxyStats
} }
} }
@@ -166,6 +168,7 @@ func (m *serverMetrics) AddTrafficIn(name string, _ string, trafficBytes int64)
proxyStats, ok := m.info.ProxyStatistics[name] proxyStats, ok := m.info.ProxyStatistics[name]
if ok { if ok {
proxyStats.TrafficIn.Inc(trafficBytes) proxyStats.TrafficIn.Inc(trafficBytes)
m.info.ProxyStatistics[name] = proxyStats
} }
} }
@@ -178,6 +181,7 @@ func (m *serverMetrics) AddTrafficOut(name string, _ string, trafficBytes int64)
proxyStats, ok := m.info.ProxyStatistics[name] proxyStats, ok := m.info.ProxyStatistics[name]
if ok { if ok {
proxyStats.TrafficOut.Inc(trafficBytes) proxyStats.TrafficOut.Inc(trafficBytes)
m.info.ProxyStatistics[name] = proxyStats
} }
} }
@@ -199,25 +203,6 @@ func (m *serverMetrics) GetServer() *ServerStats {
return s return s
} }
func toProxyStats(name string, proxyStats *ProxyStatistics) *ProxyStats {
ps := &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() {
ps.LastStartTime = proxyStats.LastStartTime.Format("01-02 15:04:05")
}
if !proxyStats.LastCloseTime.IsZero() {
ps.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
}
return ps
}
func (m *serverMetrics) GetProxiesByType(proxyType string) []*ProxyStats { func (m *serverMetrics) GetProxiesByType(proxyType string) []*ProxyStats {
res := make([]*ProxyStats, 0) res := make([]*ProxyStats, 0)
m.mu.Lock() m.mu.Lock()
@@ -227,7 +212,23 @@ func (m *serverMetrics) GetProxiesByType(proxyType string) []*ProxyStats {
if proxyStats.ProxyType != proxyType { if proxyStats.ProxyType != proxyType {
continue continue
} }
res = append(res, toProxyStats(name, proxyStats))
ps := &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() {
ps.LastStartTime = proxyStats.LastStartTime.Format("01-02 15:04:05")
}
if !proxyStats.LastCloseTime.IsZero() {
ps.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
}
res = append(res, ps)
} }
return res return res
} }
@@ -236,9 +237,31 @@ func (m *serverMetrics) GetProxiesByTypeAndName(proxyType string, proxyName stri
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
proxyStats, ok := m.info.ProxyStatistics[proxyName] for name, proxyStats := range m.info.ProxyStatistics {
if ok && proxyStats.ProxyType == proxyType { if proxyStats.ProxyType != proxyType {
res = toProxyStats(proxyName, proxyStats) continue
}
if name != proxyName {
continue
}
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 return
} }
@@ -249,7 +272,21 @@ func (m *serverMetrics) GetProxyByName(proxyName string) (res *ProxyStats) {
proxyStats, ok := m.info.ProxyStatistics[proxyName] proxyStats, ok := m.info.ProxyStatistics[proxyName]
if ok { 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 return
} }

View File

@@ -61,7 +61,7 @@ var msgTypeMap = map[byte]any{
TypeNatHoleReport: NatHoleReport{}, TypeNatHoleReport: NatHoleReport{},
} }
var TypeNameNatHoleResp = reflect.TypeFor[NatHoleResp]().Name() var TypeNameNatHoleResp = reflect.TypeOf(&NatHoleResp{}).Elem().Name()
type ClientSpec struct { type ClientSpec struct {
// Due to the support of VirtualClient, frps needs to know the client type in order to // Due to the support of VirtualClient, frps needs to know the client type in order to

View File

@@ -16,8 +16,9 @@ func StripUserPrefix(user, name string) string {
if user == "" { if user == "" {
return name 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 return name
} }

View File

@@ -151,7 +151,7 @@ func getBehaviorScoresByMode(mode int, defaultScore int) []*BehaviorScore {
func getBehaviorScoresByMode2(mode int, senderScore, receiverScore int) []*BehaviorScore { func getBehaviorScoresByMode2(mode int, senderScore, receiverScore int) []*BehaviorScore {
behaviors := getBehaviorByMode(mode) behaviors := getBehaviorByMode(mode)
scores := make([]*BehaviorScore, 0, len(behaviors)) scores := make([]*BehaviorScore, 0, len(behaviors))
for i := range behaviors { for i := 0; i < len(behaviors); i++ {
score := receiverScore score := receiverScore
if behaviors[i].A.Role == DetectRoleSender { if behaviors[i].A.Role == DetectRoleSender {
score = senderScore score = senderScore

View File

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

View File

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

View File

@@ -298,13 +298,11 @@ func waitDetectMessage(
n, raddr, err := conn.ReadFromUDP(buf) n, raddr, err := conn.ReadFromUDP(buf)
_ = conn.SetReadDeadline(time.Time{}) _ = conn.SetReadDeadline(time.Time{})
if err != nil { if err != nil {
pool.PutBuf(buf)
return nil, err return nil, err
} }
xl.Debugf("get udp message local %s, from %s", conn.LocalAddr(), raddr) xl.Debugf("get udp message local %s, from %s", conn.LocalAddr(), raddr)
var m msg.NatHoleSid var m msg.NatHoleSid
if err := DecodeMessageInto(buf[:n], key, &m); err != nil { if err := DecodeMessageInto(buf[:n], key, &m); err != nil {
pool.PutBuf(buf)
xl.Warnf("decode sid message error: %v", err) xl.Warnf("decode sid message error: %v", err)
continue continue
} }
@@ -410,7 +408,7 @@ func sendSidMessageToRandomPorts(
xl := xlog.FromContextSafe(ctx) xl := xlog.FromContextSafe(ctx)
used := sets.New[int]() used := sets.New[int]()
getUnusedPort := func() int { getUnusedPort := func() int {
for range 10 { for i := 0; i < 10; i++ {
port := rand.IntN(65535-1024) + 1024 port := rand.IntN(65535-1024) + 1024
if !used.Has(port) { if !used.Has(port) {
used.Insert(port) used.Insert(port)
@@ -420,7 +418,7 @@ func sendSidMessageToRandomPorts(
return 0 return 0
} }
for range count { for i := 0; i < count; i++ {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return

View File

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

View File

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

View File

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

View File

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

View File

@@ -24,7 +24,6 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"reflect" "reflect"
"slices"
"strings" "strings"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
@@ -65,7 +64,12 @@ func (p *httpPlugin) Name() string {
} }
func (p *httpPlugin) IsSupport(op string) bool { 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) { func (p *httpPlugin) Handle(ctx context.Context, op string, content any) (*Response, any, error) {

View File

@@ -153,7 +153,10 @@ func (p *VirtualNetPlugin) run() {
// Exponential backoff: 60s, 120s, 240s, 300s (capped) // Exponential backoff: 60s, 120s, 240s, 300s (capped)
baseDelay := 60 * time.Second baseDelay := 60 * time.Second
reconnectDelay = min(baseDelay*time.Duration(1<<uint(p.consecutiveErrors-1)), 300*time.Second) reconnectDelay = baseDelay * time.Duration(1<<uint(p.consecutiveErrors-1))
if reconnectDelay > 300*time.Second {
reconnectDelay = 300 * time.Second
}
} else { } else {
// Reset consecutive errors on successful connection // Reset consecutive errors on successful connection
if p.consecutiveErrors > 0 { if p.consecutiveErrors > 0 {

View File

@@ -16,7 +16,6 @@ package featuregate
import ( import (
"fmt" "fmt"
"maps"
"sort" "sort"
"strings" "strings"
"sync" "sync"
@@ -93,7 +92,10 @@ type featureGate struct {
// NewFeatureGate creates a new feature gate with the default features // NewFeatureGate creates a new feature gate with the default features
func NewFeatureGate() MutableFeatureGate { func NewFeatureGate() MutableFeatureGate {
known := maps.Clone(defaultFeatures) known := map[Feature]FeatureSpec{}
for k, v := range defaultFeatures {
known[k] = v
}
f := &featureGate{} f := &featureGate{}
f.known.Store(known) f.known.Store(known)
@@ -107,8 +109,14 @@ func (f *featureGate) SetFromMap(m map[string]bool) error {
defer f.lock.Unlock() defer f.lock.Unlock()
// Copy existing state // Copy existing state
known := maps.Clone(f.known.Load().(map[Feature]FeatureSpec)) known := map[Feature]FeatureSpec{}
enabled := maps.Clone(f.enabled.Load().(map[Feature]bool)) for k, v := range f.known.Load().(map[Feature]FeatureSpec) {
known[k] = v
}
enabled := map[Feature]bool{}
for k, v := range f.enabled.Load().(map[Feature]bool) {
enabled[k] = v
}
// Apply the new settings // Apply the new settings
for k, v := range m { for k, v := range m {
@@ -139,7 +147,10 @@ func (f *featureGate) Add(features map[Feature]FeatureSpec) error {
} }
// Copy existing state // Copy existing state
known := maps.Clone(f.known.Load().(map[Feature]FeatureSpec)) known := map[Feature]FeatureSpec{}
for k, v := range f.known.Load().(map[Feature]FeatureSpec) {
known[k] = v
}
// Add new features // Add new features
for name, spec := range features { for name, spec := range features {

View File

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

View File

@@ -20,7 +20,6 @@ import (
"crypto/tls" "crypto/tls"
"crypto/x509" "crypto/x509"
"encoding/pem" "encoding/pem"
"fmt"
"math/big" "math/big"
"os" "os"
"time" "time"
@@ -86,9 +85,7 @@ func newCertPool(caPath string) (*x509.CertPool, error) {
return nil, err 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 return pool, nil
} }

View File

@@ -89,11 +89,11 @@ func ParseBasicAuth(auth string) (username, password string, ok bool) {
return return
} }
cs := string(c) cs := string(c)
before, after, found := strings.Cut(cs, ":") s := strings.IndexByte(cs, ':')
if !found { if s < 0 {
return return
} }
return before, after, true return cs[:s], cs[s+1:], true
} }
func BasicAuth(username, passwd string) string { func BasicAuth(username, passwd string) string {

View File

@@ -86,7 +86,11 @@ func (c *FakeUDPConn) Read(b []byte) (n int, err error) {
c.lastActive = time.Now() c.lastActive = time.Now()
c.mu.Unlock() c.mu.Unlock()
n = min(len(b), len(content)) if len(b) < len(content) {
n = len(b)
} else {
n = len(content)
}
copy(b, content) copy(b, content)
return n, nil return n, nil
} }
@@ -164,15 +168,11 @@ func ListenUDP(bindAddr string, bindPort int) (l *UDPListener, err error) {
return l, err return l, err
} }
readConn, err := net.ListenUDP("udp", udpAddr) readConn, err := net.ListenUDP("udp", udpAddr)
if err != nil {
return l, err
}
l = &UDPListener{ l = &UDPListener{
addr: udpAddr, addr: udpAddr,
acceptCh: make(chan net.Conn), acceptCh: make(chan net.Conn),
writeCh: make(chan *UDPPacket, 1000), writeCh: make(chan *UDPPacket, 1000),
readConn: readConn,
fakeConns: make(map[string]*FakeUDPConn), fakeConns: make(map[string]*FakeUDPConn),
} }

View File

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

View File

@@ -68,8 +68,8 @@ func ParseRangeNumbers(rangeStr string) (numbers []int64, err error) {
rangeStr = strings.TrimSpace(rangeStr) rangeStr = strings.TrimSpace(rangeStr)
numbers = make([]int64, 0) numbers = make([]int64, 0)
// e.g. 1000-2000,2001,2002,3000-4000 // e.g. 1000-2000,2001,2002,3000-4000
numRanges := strings.SplitSeq(rangeStr, ",") numRanges := strings.Split(rangeStr, ",")
for numRangeStr := range numRanges { for _, numRangeStr := range numRanges {
// 1000-2000 or 2001 // 1000-2000 or 2001
numArray := strings.Split(numRangeStr, "-") numArray := strings.Split(numRangeStr, "-")
// length: only 1 or 2 is correct // length: only 1 or 2 is correct

View File

@@ -266,13 +266,31 @@ func (rp *HTTPReverseProxy) connectHandler(rw http.ResponseWriter, req *http.Req
go libio.Join(remote, client) 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 { func (rp *HTTPReverseProxy) injectRequestInfoToCtx(req *http.Request) *http.Request {
user := "" user := ""
// If url host isn't empty, it's a proxy request. Get http user from Proxy-Authorization header. // If url host isn't empty, it's a proxy request. Get http user from Proxy-Authorization header.
if req.URL.Host != "" { if req.URL.Host != "" {
proxyAuth := req.Header.Get("Proxy-Authorization") proxyAuth := req.Header.Get("Proxy-Authorization")
if proxyAuth != "" { if proxyAuth != "" {
user, _, _ = httppkg.ParseBasicAuth(proxyAuth) user, _, _ = parseBasicAuth(proxyAuth)
} }
} }
if user == "" { if user == "" {

View File

@@ -63,12 +63,11 @@ func (l *Logger) AddPrefix(prefix LogPrefix) *Logger {
if prefix.Priority <= 0 { if prefix.Priority <= 0 {
prefix.Priority = 10 prefix.Priority = 10
} }
for i, p := range l.prefixes { for _, p := range l.prefixes {
if p.Name == prefix.Name { if p.Name == prefix.Name {
found = true 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 { if !found {

View File

@@ -95,33 +95,20 @@ func (cm *ControlManager) Close() error {
return nil 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 { 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 // other components can use this to communicate with client
msgTransporter transport.MessageTransporter 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. // It provides a channel for sending messages, and you can register handlers to process messages based on their respective types.
msgDispatcher *msg.Dispatcher msgDispatcher *msg.Dispatcher
// login message
loginMsg *msg.Login
// control connection
conn net.Conn
// work connections // work connections
workConnCh chan net.Conn workConnCh chan net.Conn
@@ -152,34 +145,61 @@ type Control struct {
mu sync.RWMutex mu sync.RWMutex
// Server configuration information
serverCfg *v1.ServerConfig
clientRegistry *registry.ClientRegistry
xl *xlog.Logger xl *xlog.Logger
ctx context.Context ctx context.Context
doneCh chan struct{} 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 := min(sessionCtx.LoginMsg.PoolCount, int(sessionCtx.ServerCfg.Transport.MaxPoolCount)) func NewControl(
ctx context.Context,
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{ ctl := &Control{
sessionCtx: sessionCtx, rc: rc,
workConnCh: make(chan net.Conn, poolCount+10), pxyManager: pxyManager,
proxies: make(map[string]proxy.Proxy), pluginManager: pluginManager,
poolCount: poolCount, authVerifier: authVerifier,
portsUsedNum: 0, encryptionKey: encryptionKey,
runID: sessionCtx.LoginMsg.RunID, conn: ctlConn,
xl: xlog.FromContextSafe(ctx), loginMsg: loginMsg,
ctx: ctx, workConnCh: make(chan net.Conn, poolCount+10),
doneCh: make(chan struct{}), proxies: make(map[string]proxy.Proxy),
poolCount: poolCount,
portsUsedNum: 0,
runID: loginMsg.RunID,
serverCfg: serverCfg,
xl: xlog.FromContextSafe(ctx),
ctx: ctx,
doneCh: make(chan struct{}),
} }
ctl.lastPing.Store(time.Now()) 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 { if err != nil {
return nil, err return nil, err
} }
ctl.msgDispatcher = msg.NewDispatcher(cryptoRW) ctl.msgDispatcher = msg.NewDispatcher(cryptoRW)
} else { } else {
ctl.msgDispatcher = msg.NewDispatcher(sessionCtx.Conn) ctl.msgDispatcher = msg.NewDispatcher(ctl.conn)
} }
ctl.registerMsgHandlers() ctl.registerMsgHandlers()
ctl.msgTransporter = transport.NewMessageTransporter(ctl.msgDispatcher) ctl.msgTransporter = transport.NewMessageTransporter(ctl.msgDispatcher)
@@ -193,7 +213,7 @@ func (ctl *Control) Start() {
RunID: ctl.runID, RunID: ctl.runID,
Error: "", Error: "",
} }
_ = msg.WriteMsg(ctl.sessionCtx.Conn, loginRespMsg) _ = msg.WriteMsg(ctl.conn, loginRespMsg)
go func() { go func() {
for i := 0; i < ctl.poolCount; i++ { for i := 0; i < ctl.poolCount; i++ {
@@ -205,7 +225,7 @@ func (ctl *Control) Start() {
} }
func (ctl *Control) Close() error { func (ctl *Control) Close() error {
ctl.sessionCtx.Conn.Close() ctl.conn.Close()
return nil return nil
} }
@@ -213,7 +233,7 @@ func (ctl *Control) Replaced(newCtl *Control) {
xl := ctl.xl xl := ctl.xl
xl.Infof("replaced by client [%s]", newCtl.runID) xl.Infof("replaced by client [%s]", newCtl.runID)
ctl.runID = "" ctl.runID = ""
ctl.sessionCtx.Conn.Close() ctl.conn.Close()
} }
func (ctl *Control) RegisterWorkConn(conn net.Conn) error { func (ctl *Control) RegisterWorkConn(conn net.Conn) error {
@@ -271,7 +291,7 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
return 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") err = fmt.Errorf("timeout trying to get work connection")
xl.Warnf("%v", err) xl.Warnf("%v", err)
return return
@@ -284,15 +304,15 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
} }
func (ctl *Control) heartbeatWorker() { func (ctl *Control) heartbeatWorker() {
if ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout <= 0 { if ctl.serverCfg.Transport.HeartbeatTimeout <= 0 {
return return
} }
xl := ctl.xl xl := ctl.xl
go wait.Until(func() { 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") xl.Warnf("heartbeat timeout")
ctl.sessionCtx.Conn.Close() ctl.conn.Close()
return return
} }
}, time.Second, ctl.doneCh) }, time.Second, ctl.doneCh)
@@ -303,30 +323,6 @@ func (ctl *Control) WaitClosed() {
<-ctl.doneCh <-ctl.doneCh
} }
func (ctl *Control) loginUserInfo() plugin.UserInfo {
return plugin.UserInfo{
User: ctl.sessionCtx.LoginMsg.User,
Metas: ctl.sessionCtx.LoginMsg.Metas,
RunID: ctl.sessionCtx.LoginMsg.RunID,
}
}
func (ctl *Control) closeProxy(pxy proxy.Proxy) {
pxy.Close()
ctl.sessionCtx.PxyManager.Del(pxy.GetName())
metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
notifyContent := &plugin.CloseProxyContent{
User: ctl.loginUserInfo(),
CloseProxy: msg.CloseProxy{
ProxyName: pxy.GetName(),
},
}
go func() {
_ = ctl.sessionCtx.PluginManager.CloseProxy(notifyContent)
}()
}
func (ctl *Control) worker() { func (ctl *Control) worker() {
xl := ctl.xl xl := ctl.xl
@@ -334,23 +330,38 @@ func (ctl *Control) worker() {
go ctl.msgDispatcher.Run() go ctl.msgDispatcher.Run()
<-ctl.msgDispatcher.Done() <-ctl.msgDispatcher.Done()
ctl.sessionCtx.Conn.Close() ctl.conn.Close()
ctl.mu.Lock() ctl.mu.Lock()
defer ctl.mu.Unlock()
close(ctl.workConnCh) close(ctl.workConnCh)
for workConn := range ctl.workConnCh { for workConn := range ctl.workConnCh {
workConn.Close() workConn.Close()
} }
proxies := ctl.proxies
ctl.proxies = make(map[string]proxy.Proxy)
ctl.mu.Unlock()
for _, pxy := range proxies { for _, pxy := range ctl.proxies {
ctl.closeProxy(pxy) pxy.Close()
ctl.pxyManager.Del(pxy.GetName())
metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
notifyContent := &plugin.CloseProxyContent{
User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
RunID: ctl.loginMsg.RunID,
},
CloseProxy: msg.CloseProxy{
ProxyName: pxy.GetName(),
},
}
go func() {
_ = ctl.pluginManager.CloseProxy(notifyContent)
}()
} }
metrics.Server.CloseClient() metrics.Server.CloseClient()
ctl.sessionCtx.ClientRegistry.MarkOfflineByRunID(ctl.runID) ctl.clientRegistry.MarkOfflineByRunID(ctl.runID)
xl.Infof("client exit success") xl.Infof("client exit success")
close(ctl.doneCh) close(ctl.doneCh)
} }
@@ -369,11 +380,15 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
inMsg := m.(*msg.NewProxy) inMsg := m.(*msg.NewProxy)
content := &plugin.NewProxyContent{ content := &plugin.NewProxyContent{
User: ctl.loginUserInfo(), User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
RunID: ctl.loginMsg.RunID,
},
NewProxy: *inMsg, NewProxy: *inMsg,
} }
var remoteAddr string var remoteAddr string
retContent, err := ctl.sessionCtx.PluginManager.NewProxy(content) retContent, err := ctl.pluginManager.NewProxy(content)
if err == nil { if err == nil {
inMsg = &retContent.NewProxy inMsg = &retContent.NewProxy
remoteAddr, err = ctl.RegisterProxy(inMsg) remoteAddr, err = ctl.RegisterProxy(inMsg)
@@ -386,15 +401,15 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
if err != nil { if err != nil {
xl.Warnf("new proxy [%s] type [%s] error: %v", inMsg.ProxyName, inMsg.ProxyType, err) 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), 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 { } else {
resp.RemoteAddr = remoteAddr resp.RemoteAddr = remoteAddr
xl.Infof("new proxy [%s] type [%s] success", inMsg.ProxyName, inMsg.ProxyType) xl.Infof("new proxy [%s] type [%s] success", inMsg.ProxyName, inMsg.ProxyType)
clientID := ctl.sessionCtx.LoginMsg.ClientID clientID := ctl.loginMsg.ClientID
if 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) _ = ctl.msgDispatcher.Send(resp)
} }
@@ -404,18 +419,22 @@ func (ctl *Control) handlePing(m msg.Message) {
inMsg := m.(*msg.Ping) inMsg := m.(*msg.Ping)
content := &plugin.PingContent{ content := &plugin.PingContent{
User: ctl.loginUserInfo(), User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
RunID: ctl.loginMsg.RunID,
},
Ping: *inMsg, Ping: *inMsg,
} }
retContent, err := ctl.sessionCtx.PluginManager.Ping(content) retContent, err := ctl.pluginManager.Ping(content)
if err == nil { if err == nil {
inMsg = &retContent.Ping inMsg = &retContent.Ping
err = ctl.sessionCtx.AuthVerifier.VerifyPing(inMsg) err = ctl.authVerifier.VerifyPing(inMsg)
} }
if err != nil { if err != nil {
xl.Warnf("received invalid ping: %v", err) xl.Warnf("received invalid ping: %v", err)
_ = ctl.msgDispatcher.Send(&msg.Pong{ _ = 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 return
} }
@@ -426,17 +445,17 @@ func (ctl *Control) handlePing(m msg.Message) {
func (ctl *Control) handleNatHoleVisitor(m msg.Message) { func (ctl *Control) handleNatHoleVisitor(m msg.Message) {
inMsg := m.(*msg.NatHoleVisitor) 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) { func (ctl *Control) handleNatHoleClient(m msg.Message) {
inMsg := m.(*msg.NatHoleClient) 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) { func (ctl *Control) handleNatHoleReport(m msg.Message) {
inMsg := m.(*msg.NatHoleReport) inMsg := m.(*msg.NatHoleReport)
ctl.sessionCtx.RC.NatHoleController.HandleReport(inMsg) ctl.rc.NatHoleController.HandleReport(inMsg)
} }
func (ctl *Control) handleCloseProxy(m msg.Message) { func (ctl *Control) handleCloseProxy(m msg.Message) {
@@ -449,15 +468,15 @@ func (ctl *Control) handleCloseProxy(m msg.Message) {
func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err error) { func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err error) {
var pxyConf v1.ProxyConfigurer var pxyConf v1.ProxyConfigurer
// Load configures from NewProxy message and validate. // 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 { if err != nil {
return return
} }
// User info // User info
userInfo := plugin.UserInfo{ userInfo := plugin.UserInfo{
User: ctl.sessionCtx.LoginMsg.User, User: ctl.loginMsg.User,
Metas: ctl.sessionCtx.LoginMsg.Metas, Metas: ctl.loginMsg.Metas,
RunID: ctl.runID, RunID: ctl.runID,
} }
@@ -465,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. // In fact, it creates different proxies based on the proxy type. We just call run() here.
pxy, err := proxy.NewProxy(ctl.ctx, &proxy.Options{ pxy, err := proxy.NewProxy(ctl.ctx, &proxy.Options{
UserInfo: userInfo, UserInfo: userInfo,
LoginMsg: ctl.sessionCtx.LoginMsg, LoginMsg: ctl.loginMsg,
PoolCount: ctl.poolCount, PoolCount: ctl.poolCount,
ResourceController: ctl.sessionCtx.RC, ResourceController: ctl.rc,
GetWorkConnFn: ctl.GetWorkConn, GetWorkConnFn: ctl.GetWorkConn,
Configurer: pxyConf, Configurer: pxyConf,
ServerCfg: ctl.sessionCtx.ServerCfg, ServerCfg: ctl.serverCfg,
EncryptionKey: ctl.sessionCtx.EncryptionKey, EncryptionKey: ctl.encryptionKey,
}) })
if err != nil { if err != nil {
return remoteAddr, err return remoteAddr, err
} }
// Check ports used number in each client // Check ports used number in each client
if ctl.sessionCtx.ServerCfg.MaxPortsPerClient > 0 { if ctl.serverCfg.MaxPortsPerClient > 0 {
ctl.mu.Lock() 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() ctl.mu.Unlock()
err = fmt.Errorf("exceed the max_ports_per_client") err = fmt.Errorf("exceed the max_ports_per_client")
return return
@@ -497,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) err = fmt.Errorf("proxy [%s] already exists", pxyMsg.ProxyName)
return return
} }
@@ -512,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 { if err != nil {
return return
} }
@@ -531,12 +550,28 @@ func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
return return
} }
if ctl.sessionCtx.ServerCfg.MaxPortsPerClient > 0 { if ctl.serverCfg.MaxPortsPerClient > 0 {
ctl.portsUsedNum -= pxy.GetUsedPortsNum() ctl.portsUsedNum -= pxy.GetUsedPortsNum()
} }
pxy.Close()
ctl.pxyManager.Del(pxy.GetName())
delete(ctl.proxies, closeMsg.ProxyName) delete(ctl.proxies, closeMsg.ProxyName)
ctl.mu.Unlock() ctl.mu.Unlock()
ctl.closeProxy(pxy) metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
notifyContent := &plugin.CloseProxyContent{
User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
RunID: ctl.loginMsg.RunID,
},
CloseProxy: msg.CloseProxy{
ProxyName: pxy.GetName(),
},
}
go func() {
_ = ctl.pluginManager.CloseProxy(notifyContent)
}()
return return
} }

View File

@@ -100,9 +100,8 @@ func (tg *TCPGroup) Listen(proxyName string, group string, groupKey string, addr
if err != nil { if err != nil {
return 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 { if errRet != nil {
tg.ctl.portManager.Release(realPort)
err = errRet err = errRet
return return
} }

View File

@@ -31,7 +31,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.HTTPProxyConfig](), NewHTTPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.HTTPProxyConfig{}), NewHTTPProxy)
} }
type HTTPProxy struct { type HTTPProxy struct {
@@ -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) addrs := make([]string, 0)
for _, domain := range domains { for _, domain := range pxy.cfg.CustomDomains {
if domain == "" {
continue
}
routeConfig.Domain = domain routeConfig.Domain = domain
for _, location := range locations { for _, location := range locations {
routeConfig.Location = location routeConfig.Location = location
tmpRouteConfig := routeConfig tmpRouteConfig := routeConfig
// handle group // handle group
@@ -90,6 +93,40 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
if err != nil { if err != nil {
return 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.closeFuncs = append(pxy.closeFuncs, func() {
pxy.rc.HTTPGroupCtl.UnRegister(pxy.name, pxy.cfg.LoadBalancer.Group, tmpRouteConfig) 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) 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]", 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) 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) rwc, err = libio.WithEncryption(rwc, pxy.encryptionKey)
if err != nil { if err != nil {
xl.Errorf("create encryption stream error: %v", err) xl.Errorf("create encryption stream error: %v", err)
tmpConn.Close()
return return
} }
} }

View File

@@ -25,7 +25,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.HTTPSProxyConfig](), NewHTTPSProxy) RegisterProxyFactory(reflect.TypeOf(&v1.HTTPSProxyConfig{}), NewHTTPSProxy)
} }
type HTTPSProxy struct { type HTTPSProxy struct {
@@ -53,10 +53,23 @@ func (pxy *HTTPSProxy) Run() (remoteAddr string, err error) {
pxy.Close() pxy.Close()
} }
}() }()
domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
addrs := make([]string, 0) 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) l, err := pxy.listenForDomain(routeConfig, domain)
if err != nil { if err != nil {
return "", err return "", err

View File

@@ -150,7 +150,7 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
dstAddr, dstPortStr, _ = net.SplitHostPort(dst.String()) dstAddr, dstPortStr, _ = net.SplitHostPort(dst.String())
dstPort, _ = strconv.ParseUint(dstPortStr, 10, 16) dstPort, _ = strconv.ParseUint(dstPortStr, 10, 16)
} }
err = msg.WriteMsg(workConn, &msg.StartWorkConn{ err := msg.WriteMsg(workConn, &msg.StartWorkConn{
ProxyName: pxy.GetName(), ProxyName: pxy.GetName(),
SrcAddr: srcAddr, SrcAddr: srcAddr,
SrcPort: uint16(srcPort), SrcPort: uint16(srcPort),
@@ -161,7 +161,6 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
if err != nil { if err != nil {
xl.Warnf("failed to send message to work connection from pool: %v, times: %d", err, i) xl.Warnf("failed to send message to work connection from pool: %v, times: %d", err, i)
workConn.Close() workConn.Close()
workConn = nil
} else { } else {
break break
} }
@@ -174,36 +173,6 @@ func (pxy *BaseProxy) GetWorkConnFromPool(src, dst net.Addr) (workConn net.Conn,
return 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. // startCommonTCPListenersHandler start a goroutine handler for each listener.
func (pxy *BaseProxy) startCommonTCPListenersHandler() { func (pxy *BaseProxy) startCommonTCPListenersHandler() {
xl := xlog.FromContextSafe(pxy.ctx) xl := xlog.FromContextSafe(pxy.ctx)

View File

@@ -21,7 +21,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.STCPProxyConfig](), NewSTCPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.STCPProxyConfig{}), NewSTCPProxy)
} }
type STCPProxy struct { type STCPProxy struct {
@@ -41,7 +41,21 @@ func NewSTCPProxy(baseProxy *BaseProxy) Proxy {
} }
func (pxy *STCPProxy) Run() (remoteAddr string, err error) { 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 return
} }

View File

@@ -21,7 +21,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.SUDPProxyConfig](), NewSUDPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.SUDPProxyConfig{}), NewSUDPProxy)
} }
type SUDPProxy struct { type SUDPProxy struct {
@@ -41,7 +41,21 @@ func NewSUDPProxy(baseProxy *BaseProxy) Proxy {
} }
func (pxy *SUDPProxy) Run() (remoteAddr string, err error) { 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 return
} }

View File

@@ -24,7 +24,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.TCPProxyConfig](), NewTCPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.TCPProxyConfig{}), NewTCPProxy)
} }
type TCPProxy struct { type TCPProxy struct {

View File

@@ -26,7 +26,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.TCPMuxProxyConfig](), NewTCPMuxProxy) RegisterProxyFactory(reflect.TypeOf(&v1.TCPMuxProxyConfig{}), NewTCPMuxProxy)
} }
type TCPMuxProxy struct { type TCPMuxProxy struct {
@@ -72,16 +72,26 @@ func (pxy *TCPMuxProxy) httpConnectListen(
} }
func (pxy *TCPMuxProxy) httpConnectRun() (remoteAddr string, err error) { func (pxy *TCPMuxProxy) httpConnectRun() (remoteAddr string, err error) {
domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
addrs := make([]string, 0) 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) addrs, err = pxy.httpConnectListen(domain, pxy.cfg.RouteByHTTPUser, pxy.cfg.HTTPUser, pxy.cfg.HTTPPassword, addrs)
if err != nil { if err != nil {
return "", err 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() pxy.startCommonTCPListenersHandler()
remoteAddr = strings.Join(addrs, ",") remoteAddr = strings.Join(addrs, ",")
return remoteAddr, err return remoteAddr, err

View File

@@ -35,7 +35,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.UDPProxyConfig](), NewUDPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.UDPProxyConfig{}), NewUDPProxy)
} }
type UDPProxy struct { type UDPProxy struct {

View File

@@ -24,7 +24,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeFor[*v1.XTCPProxyConfig](), NewXTCPProxy) RegisterProxyFactory(reflect.TypeOf(&v1.XTCPProxyConfig{}), NewXTCPProxy)
} }
type XTCPProxy struct { type XTCPProxy struct {

View File

@@ -193,7 +193,7 @@ func NewService(cfg *v1.ServerConfig) (*Service, error) {
if err != nil { if err != nil {
return nil, fmt.Errorf("create vhost tcpMuxer error, %v", err) return nil, fmt.Errorf("create vhost tcpMuxer error, %v", err)
} }
log.Infof("tcpmux httpconnect multiplexer listen on %s, passthrough: %v", address, cfg.TCPMuxPassthrough) log.Infof("tcpmux httpconnect multiplexer listen on %s, passthough: %v", address, cfg.TCPMuxPassthrough)
} }
// Init all plugins // Init all plugins
@@ -604,18 +604,8 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, inter
return err 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 { if err != nil {
xl.Warnf("create new controller error: %v", err) xl.Warnf("create new controller error: %v", err)
// don't return detailed errors to client // don't return detailed errors to client
@@ -636,6 +626,7 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, inter
ctl.Close() ctl.Close()
return fmt.Errorf("client_id [%s] for user [%s] is already online", loginMsg.ClientID, loginMsg.User) return fmt.Errorf("client_id [%s] for user [%s] is already online", loginMsg.ClientID, loginMsg.User)
} }
ctl.clientRegistry = svr.clientRegistry
ctl.Start() ctl.Start()
@@ -661,9 +652,9 @@ func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn)
// server plugin hook // server plugin hook
content := &plugin.NewWorkConnContent{ content := &plugin.NewWorkConnContent{
User: plugin.UserInfo{ 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, NewWorkConn: *newMsg,
} }
@@ -671,7 +662,7 @@ func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn)
if err == nil { if err == nil {
newMsg = &retContent.NewWorkConn newMsg = &retContent.NewWorkConn
// Check auth. // Check auth.
err = ctl.sessionCtx.AuthVerifier.VerifyNewWorkConn(newMsg) err = ctl.authVerifier.VerifyNewWorkConn(newMsg)
} }
if err != nil { if err != nil {
xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID) 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 { if !exist {
return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID) 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, return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
newMsg.UseEncryption, newMsg.UseCompression, visitorUser) newMsg.UseEncryption, newMsg.UseCompression, visitorUser)

View File

@@ -2,10 +2,8 @@ package framework
import ( import (
"fmt" "fmt"
"maps"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"time" "time"
flog "github.com/fatedier/frp/pkg/util/log" flog "github.com/fatedier/frp/pkg/util/log"
@@ -15,12 +13,16 @@ import (
// RunProcesses run multiple processes from templates. // RunProcesses run multiple processes from templates.
// The first template should always be frps. // The first template should always be frps.
func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []string) ([]*process.Process, []*process.Process) { func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []string) ([]*process.Process, []*process.Process) {
templates := slices.Concat(serverTemplates, clientTemplates) templates := make([]string, 0, len(serverTemplates)+len(clientTemplates))
templates = append(templates, serverTemplates...)
templates = append(templates, clientTemplates...)
outs, ports, err := f.RenderTemplates(templates) outs, ports, err := f.RenderTemplates(templates)
ExpectNoError(err) ExpectNoError(err)
ExpectTrue(len(templates) > 0) ExpectTrue(len(templates) > 0)
maps.Copy(f.usedPorts, ports) for name, port := range ports {
f.usedPorts[name] = port
}
currentServerProcesses := make([]*process.Process, 0, len(serverTemplates)) currentServerProcesses := make([]*process.Process, 0, len(serverTemplates))
for i := range serverTemplates { for i := range serverTemplates {

View File

@@ -26,8 +26,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
proxyType := t proxyType := t
ginkgo.It(fmt.Sprintf("Expose a %s echo server", strings.ToUpper(proxyType)), func() { ginkgo.It(fmt.Sprintf("Expose a %s echo server", strings.ToUpper(proxyType)), func() {
serverConf := consts.LegacyDefaultServerConfig serverConf := consts.LegacyDefaultServerConfig
var clientConf strings.Builder clientConf := consts.LegacyDefaultClientConfig
clientConf.WriteString(consts.LegacyDefaultClientConfig)
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -79,10 +78,10 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
@@ -103,8 +102,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
vhost_http_port = %d vhost_http_port = %d
`, vhostHTTPPort) `, vhostHTTPPort)
var clientConf strings.Builder clientConf := consts.LegacyDefaultClientConfig
clientConf.WriteString(consts.LegacyDefaultClientConfig)
getProxyConf := func(proxyName string, customDomains string, extra string) string { getProxyConf := func(proxyName string, customDomains string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
@@ -149,13 +147,13 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
if tests[i].customDomains == "" { if tests[i].customDomains == "" {
tests[i].customDomains = test.proxyName + ".example.com" tests[i].customDomains = test.proxyName + ".example.com"
} }
clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
for domain := range strings.SplitSeq(test.customDomains, ",") { for _, domain := range strings.Split(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain(test.proxyName + "-" + domain). Explain(test.proxyName + "-" + domain).
@@ -187,8 +185,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
`, vhostHTTPSPort) `, vhostHTTPSPort)
localPort := f.AllocPort() localPort := f.AllocPort()
var clientConf strings.Builder clientConf := consts.LegacyDefaultClientConfig
clientConf.WriteString(consts.LegacyDefaultClientConfig)
getProxyConf := func(proxyName string, customDomains string, extra string) string { getProxyConf := func(proxyName string, customDomains string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
[%s] [%s]
@@ -232,10 +229,10 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
if tests[i].customDomains == "" { if tests[i].customDomains == "" {
tests[i].customDomains = test.proxyName + ".example.com" tests[i].customDomains = test.proxyName + ".example.com"
} }
clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
tlsConfig, err := transport.NewServerTLSConfig("", "", "") tlsConfig, err := transport.NewServerTLSConfig("", "", "")
framework.ExpectNoError(err) framework.ExpectNoError(err)
@@ -247,7 +244,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
f.RunServer("", localServer) f.RunServer("", localServer)
for _, test := range tests { for _, test := range tests {
for domain := range strings.SplitSeq(test.customDomains, ",") { for _, domain := range strings.Split(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain(test.proxyName + "-" + domain). Explain(test.proxyName + "-" + domain).
@@ -285,12 +282,9 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
proxyType := t proxyType := t
ginkgo.It(fmt.Sprintf("Expose echo server with %s", strings.ToUpper(proxyType)), func() { ginkgo.It(fmt.Sprintf("Expose echo server with %s", strings.ToUpper(proxyType)), func() {
serverConf := consts.LegacyDefaultServerConfig serverConf := consts.LegacyDefaultServerConfig
var clientServerConf strings.Builder clientServerConf := consts.LegacyDefaultClientConfig + "\nuser = user1"
clientServerConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user1") clientVisitorConf := consts.LegacyDefaultClientConfig + "\nuser = user1"
var clientVisitorConf strings.Builder clientUser2VisitorConf := consts.LegacyDefaultClientConfig + "\nuser = user2"
clientVisitorConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user1")
var clientUser2VisitorConf strings.Builder
clientUser2VisitorConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user2")
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -406,20 +400,20 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientServerConf.WriteString(getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n") clientServerConf += getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n"
} }
for _, test := range tests { for _, test := range tests {
config := getProxyVisitorConf( config := getProxyVisitorConf(
test.proxyName, test.bindPortName, test.visitorSK, test.commonExtraConfig+"\n"+test.visitorExtraConfig, test.proxyName, test.bindPortName, test.visitorSK, test.commonExtraConfig+"\n"+test.visitorExtraConfig,
) + "\n" ) + "\n"
if test.deployUser2Client { if test.deployUser2Client {
clientUser2VisitorConf.WriteString(config) clientUser2VisitorConf += config
} else { } else {
clientVisitorConf.WriteString(config) clientVisitorConf += config
} }
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientServerConf.String(), clientVisitorConf.String(), clientUser2VisitorConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientServerConf, clientVisitorConf, clientUser2VisitorConf})
for _, test := range tests { for _, test := range tests {
timeout := time.Second timeout := time.Second
@@ -446,8 +440,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
ginkgo.Describe("TCPMUX", func() { ginkgo.Describe("TCPMUX", func() {
ginkgo.It("Type tcpmux", func() { ginkgo.It("Type tcpmux", func() {
serverConf := consts.LegacyDefaultServerConfig serverConf := consts.LegacyDefaultServerConfig
var clientConf strings.Builder clientConf := consts.LegacyDefaultClientConfig
clientConf.WriteString(consts.LegacyDefaultClientConfig)
tcpmuxHTTPConnectPortName := port.GenName("TCPMUX") tcpmuxHTTPConnectPortName := port.GenName("TCPMUX")
serverConf += fmt.Sprintf(` serverConf += fmt.Sprintf(`
@@ -490,14 +483,14 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.extraConfig) + "\n"
localServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(f.AllocPort()), streamserver.WithRespContent([]byte(test.proxyName))) localServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(f.AllocPort()), streamserver.WithRespContent([]byte(test.proxyName)))
f.RunServer(port.GenName(test.proxyName), localServer) f.RunServer(port.GenName(test.proxyName), localServer)
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
// Request without HTTP connect should get error // Request without HTTP connect should get error
framework.NewRequestExpect(f). framework.NewRequestExpect(f).

View File

@@ -28,7 +28,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
tcpPortName := port.GenName("TCP", port.WithRangePorts(10000, 11000)) tcpPortName := port.GenName("TCP", port.WithRangePorts(10000, 11000))
udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000)) udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000))
clientConf += fmt.Sprintf(` clientConf += fmt.Sprintf(`
[tcp-allowed-in-range] [tcp-allowded-in-range]
type = tcp type = tcp
local_port = {{ .%s }} local_port = {{ .%s }}
remote_port = {{ .%s }} remote_port = {{ .%s }}

View File

@@ -48,10 +48,12 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
return true return true
}) })
} }
for range 10 { for i := 0; i < 10; i++ {
wait.Go(func() { wait.Add(1)
go func() {
defer wait.Done()
expectFn() expectFn()
}) }()
} }
wait.Wait() wait.Wait()
@@ -92,7 +94,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Explain("times " + strconv.Itoa(i)).Port(remotePort).Ensure(func(resp *request.Response) bool { framework.NewRequestExpect(f).Explain("times " + strconv.Itoa(i)).Port(remotePort).Ensure(func(resp *request.Response) bool {
switch string(resp.Content) { switch string(resp.Content) {
case "foo": case "foo":
@@ -148,7 +150,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// check foo and bar is ok // check foo and bar is ok
results := []string{} results := []string{}
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
results = append(results, string(resp.Content)) results = append(results, string(resp.Content))
return true return true
@@ -159,7 +161,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// close bar server, check foo is ok // close bar server, check foo is ok
barServer.Close() barServer.Close()
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
} }
@@ -167,7 +169,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
f.RunServer("", barServer) f.RunServer("", barServer)
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
results = []string{} results = []string{}
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
results = append(results, string(resp.Content)) results = append(results, string(resp.Content))
return true return true

View File

@@ -4,7 +4,6 @@ import (
"crypto/tls" "crypto/tls"
"fmt" "fmt"
"strconv" "strconv"
"strings"
"github.com/onsi/ginkgo/v2" "github.com/onsi/ginkgo/v2"
@@ -23,8 +22,7 @@ var _ = ginkgo.Describe("[Feature: Client-Plugins]", func() {
ginkgo.Describe("UnixDomainSocket", func() { ginkgo.Describe("UnixDomainSocket", func() {
ginkgo.It("Expose a unix domain socket echo server", func() { ginkgo.It("Expose a unix domain socket echo server", func() {
serverConf := consts.LegacyDefaultServerConfig serverConf := consts.LegacyDefaultServerConfig
var clientConf strings.Builder clientConf := consts.LegacyDefaultClientConfig
clientConf.WriteString(consts.LegacyDefaultClientConfig)
getProxyConf := func(proxyName string, portName string, extra string) string { getProxyConf := func(proxyName string, portName string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
@@ -67,10 +65,10 @@ var _ = ginkgo.Describe("[Feature: Client-Plugins]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f).Port(f.PortByName(test.portName)).Ensure() framework.NewRequestExpect(f).Port(f.PortByName(test.portName)).Ensure()

View File

@@ -52,7 +52,7 @@ func (pa *Allocator) GetByName(portName string) int {
pa.mu.Lock() pa.mu.Lock()
defer pa.mu.Unlock() defer pa.mu.Unlock()
for range 20 { for i := 0; i < 20; i++ {
port := pa.getByRange(builder.rangePortFrom, builder.rangePortTo) port := pa.getByRange(builder.rangePortFrom, builder.rangePortTo)
if port == 0 { if port == 0 {
return 0 return 0

View File

@@ -75,11 +75,11 @@ func (c *TunnelClient) serveListener() {
if err != nil { if err != nil {
return return
} }
go c.handleConn(conn) go c.hanldeConn(conn)
} }
} }
func (c *TunnelClient) handleConn(conn net.Conn) { func (c *TunnelClient) hanldeConn(conn net.Conn) {
defer conn.Close() defer conn.Close()
local, err := net.Dial("tcp", c.localAddr) local, err := net.Dial("tcp", c.localAddr)
if err != nil { if err != nil {

View File

@@ -26,8 +26,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
proxyType := t proxyType := t
ginkgo.It(fmt.Sprintf("Expose a %s echo server", strings.ToUpper(proxyType)), func() { ginkgo.It(fmt.Sprintf("Expose a %s echo server", strings.ToUpper(proxyType)), func() {
serverConf := consts.DefaultServerConfig serverConf := consts.DefaultServerConfig
var clientConf strings.Builder clientConf := consts.DefaultClientConfig
clientConf.WriteString(consts.DefaultClientConfig)
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -80,10 +79,10 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
@@ -104,8 +103,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
vhostHTTPPort = %d vhostHTTPPort = %d
`, vhostHTTPPort) `, vhostHTTPPort)
var clientConf strings.Builder clientConf := consts.DefaultClientConfig
clientConf.WriteString(consts.DefaultClientConfig)
getProxyConf := func(proxyName string, customDomains string, extra string) string { getProxyConf := func(proxyName string, customDomains string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
@@ -151,13 +149,13 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
if tests[i].customDomains == "" { if tests[i].customDomains == "" {
tests[i].customDomains = fmt.Sprintf(`["%s"]`, test.proxyName+".example.com") tests[i].customDomains = fmt.Sprintf(`["%s"]`, test.proxyName+".example.com")
} }
clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
for domain := range strings.SplitSeq(test.customDomains, ",") { for _, domain := range strings.Split(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
domain = strings.TrimLeft(domain, "[\"") domain = strings.TrimLeft(domain, "[\"")
domain = strings.TrimRight(domain, "]\"") domain = strings.TrimRight(domain, "]\"")
@@ -191,8 +189,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
`, vhostHTTPSPort) `, vhostHTTPSPort)
localPort := f.AllocPort() localPort := f.AllocPort()
var clientConf strings.Builder clientConf := consts.DefaultClientConfig
clientConf.WriteString(consts.DefaultClientConfig)
getProxyConf := func(proxyName string, customDomains string, extra string) string { getProxyConf := func(proxyName string, customDomains string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
[[proxies]] [[proxies]]
@@ -237,10 +234,10 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
if tests[i].customDomains == "" { if tests[i].customDomains == "" {
tests[i].customDomains = fmt.Sprintf(`["%s"]`, test.proxyName+".example.com") tests[i].customDomains = fmt.Sprintf(`["%s"]`, test.proxyName+".example.com")
} }
clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
tlsConfig, err := transport.NewServerTLSConfig("", "", "") tlsConfig, err := transport.NewServerTLSConfig("", "", "")
framework.ExpectNoError(err) framework.ExpectNoError(err)
@@ -252,7 +249,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
f.RunServer("", localServer) f.RunServer("", localServer)
for _, test := range tests { for _, test := range tests {
for domain := range strings.SplitSeq(test.customDomains, ",") { for _, domain := range strings.Split(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
domain = strings.TrimLeft(domain, "[\"") domain = strings.TrimLeft(domain, "[\"")
domain = strings.TrimRight(domain, "]\"") domain = strings.TrimRight(domain, "]\"")
@@ -292,12 +289,9 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
proxyType := t proxyType := t
ginkgo.It(fmt.Sprintf("Expose echo server with %s", strings.ToUpper(proxyType)), func() { ginkgo.It(fmt.Sprintf("Expose echo server with %s", strings.ToUpper(proxyType)), func() {
serverConf := consts.DefaultServerConfig serverConf := consts.DefaultServerConfig
var clientServerConf strings.Builder clientServerConf := consts.DefaultClientConfig + "\nuser = \"user1\""
clientServerConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user1\"") clientVisitorConf := consts.DefaultClientConfig + "\nuser = \"user1\""
var clientVisitorConf strings.Builder clientUser2VisitorConf := consts.DefaultClientConfig + "\nuser = \"user2\""
clientVisitorConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user1\"")
var clientUser2VisitorConf strings.Builder
clientUser2VisitorConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user2\"")
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -413,20 +407,20 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientServerConf.WriteString(getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n") clientServerConf += getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n"
} }
for _, test := range tests { for _, test := range tests {
config := getProxyVisitorConf( config := getProxyVisitorConf(
test.proxyName, test.bindPortName, test.visitorSK, test.commonExtraConfig+"\n"+test.visitorExtraConfig, test.proxyName, test.bindPortName, test.visitorSK, test.commonExtraConfig+"\n"+test.visitorExtraConfig,
) + "\n" ) + "\n"
if test.deployUser2Client { if test.deployUser2Client {
clientUser2VisitorConf.WriteString(config) clientUser2VisitorConf += config
} else { } else {
clientVisitorConf.WriteString(config) clientVisitorConf += config
} }
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientServerConf.String(), clientVisitorConf.String(), clientUser2VisitorConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientServerConf, clientVisitorConf, clientUser2VisitorConf})
for _, test := range tests { for _, test := range tests {
timeout := time.Second timeout := time.Second
@@ -453,8 +447,7 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
ginkgo.Describe("TCPMUX", func() { ginkgo.Describe("TCPMUX", func() {
ginkgo.It("Type tcpmux", func() { ginkgo.It("Type tcpmux", func() {
serverConf := consts.DefaultServerConfig serverConf := consts.DefaultServerConfig
var clientConf strings.Builder clientConf := consts.DefaultClientConfig
clientConf.WriteString(consts.DefaultClientConfig)
tcpmuxHTTPConnectPortName := port.GenName("TCPMUX") tcpmuxHTTPConnectPortName := port.GenName("TCPMUX")
serverConf += fmt.Sprintf(` serverConf += fmt.Sprintf(`
@@ -498,14 +491,14 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.extraConfig) + "\n"
localServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(f.AllocPort()), streamserver.WithRespContent([]byte(test.proxyName))) localServer := streamserver.New(streamserver.TCP, streamserver.WithBindPort(f.AllocPort()), streamserver.WithRespContent([]byte(test.proxyName)))
f.RunServer(port.GenName(test.proxyName), localServer) f.RunServer(port.GenName(test.proxyName), localServer)
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
// Request without HTTP connect should get error // Request without HTTP connect should get error
framework.NewRequestExpect(f). framework.NewRequestExpect(f).

View File

@@ -33,7 +33,7 @@ var _ = ginkgo.Describe("[Feature: Server Manager]", func() {
udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000)) udpPortName := port.GenName("UDP", port.WithRangePorts(12000, 13000))
clientConf += fmt.Sprintf(` clientConf += fmt.Sprintf(`
[[proxies]] [[proxies]]
name = "tcp-allowed-in-range" name = "tcp-allowded-in-range"
type = "tcp" type = "tcp"
localPort = {{ .%s }} localPort = {{ .%s }}
remotePort = {{ .%s }} remotePort = {{ .%s }}

View File

@@ -50,10 +50,12 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
return true return true
}) })
} }
for range 10 { for i := 0; i < 10; i++ {
wait.Go(func() { wait.Add(1)
go func() {
defer wait.Done()
expectFn() expectFn()
}) }()
} }
wait.Wait() wait.Wait()
@@ -96,7 +98,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Explain("times " + strconv.Itoa(i)).Port(remotePort).Ensure(func(resp *request.Response) bool { framework.NewRequestExpect(f).Explain("times " + strconv.Itoa(i)).Port(remotePort).Ensure(func(resp *request.Response) bool {
switch string(resp.Content) { switch string(resp.Content) {
case "foo": case "foo":
@@ -161,7 +163,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain("times " + strconv.Itoa(i)). Explain("times " + strconv.Itoa(i)).
Port(vhostHTTPSPort). Port(vhostHTTPSPort).
@@ -228,7 +230,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// check foo and bar is ok // check foo and bar is ok
results := []string{} results := []string{}
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
results = append(results, string(resp.Content)) results = append(results, string(resp.Content))
return true return true
@@ -239,7 +241,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// close bar server, check foo is ok // close bar server, check foo is ok
barServer.Close() barServer.Close()
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
} }
@@ -247,7 +249,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
f.RunServer("", barServer) f.RunServer("", barServer)
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
results = []string{} results = []string{}
for range 10 { for i := 0; i < 10; i++ {
framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool { framework.NewRequestExpect(f).Port(remotePort).Ensure(validateFooBarResponse, func(resp *request.Response) bool {
results = append(results, string(resp.Content)) results = append(results, string(resp.Content))
return true return true

View File

@@ -5,7 +5,6 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"github.com/onsi/ginkgo/v2" "github.com/onsi/ginkgo/v2"
@@ -24,8 +23,7 @@ var _ = ginkgo.Describe("[Feature: Client-Plugins]", func() {
ginkgo.Describe("UnixDomainSocket", func() { ginkgo.Describe("UnixDomainSocket", func() {
ginkgo.It("Expose a unix domain socket echo server", func() { ginkgo.It("Expose a unix domain socket echo server", func() {
serverConf := consts.DefaultServerConfig serverConf := consts.DefaultServerConfig
var clientConf strings.Builder clientConf := consts.DefaultClientConfig
clientConf.WriteString(consts.DefaultClientConfig)
getProxyConf := func(proxyName string, portName string, extra string) string { getProxyConf := func(proxyName string, portName string, extra string) string {
return fmt.Sprintf(` return fmt.Sprintf(`
@@ -71,10 +69,10 @@ var _ = ginkgo.Describe("[Feature: Client-Plugins]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n") clientConf += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n"
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf.String()}) f.RunProcesses([]string{serverConf}, []string{clientConf})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f).Port(f.PortByName(test.portName)).Ensure() framework.NewRequestExpect(f).Port(f.PortByName(test.portName)).Ensure()