Compare commits

...

14 Commits

Author SHA1 Message Date
fatedier
f7aa38b303 pkg/metrics/mem: remove redundant map write-backs and optimize proxy lookup
Remove 4 redundant pointer map write-backs in OpenConnection,
CloseConnection, AddTrafficIn, and AddTrafficOut since the map stores
pointers and mutations are already visible without reassignment.

Optimize GetProxiesByTypeAndName from O(n) full map scan to O(1) direct
map lookup by proxy name.
2026-03-08 10:19:42 +08:00
fatedier
605f3bdece client/visitor: deduplicate visitor connection handshake and wrapping (#5219)
Extract two shared helpers to eliminate duplicated code across STCP,
SUDP, and XTCP visitors:

- dialRawVisitorConn: handles ConnectServer + NewVisitorConn handshake
  (auth, sign key, 10s read deadline, error check)
- wrapVisitorConn: handles encryption + pooled compression wrapping,
  returning a recycleFn for pool resource cleanup

SUDP is upgraded from WithCompression to WithCompressionFromPool,
aligning with the pooled compression used by STCP and XTCP.
2026-03-08 01:03:40 +08:00
fatedier
764a626b6e server/control: deduplicate close-proxy logic and UserInfo construction (#5218)
Extract closeProxy() helper to eliminate duplicated 4-step cleanup
sequence (Close, PxyManager.Del, metrics, plugin notify) between
worker() and CloseProxy().

Extract loginUserInfo() helper to eliminate 4 repeated plugin.UserInfo
constructions using LoginMsg fields.

Optimize worker() to snapshot and clear the proxies map under lock,
then perform cleanup outside the lock to reduce lock hold time.
2026-03-08 00:02:14 +08:00
Oleksandr Redko
c2454e7114 refactor: fix modernize lint issues (#5215) 2026-03-07 23:10:19 +08:00
fatedier
017d71717f server: introduce SessionContext to encapsulate NewControl parameters (#5217)
Replace 10 positional parameters in NewControl() with a single
SessionContext struct, matching the client-side pattern. This also
eliminates the post-construction mutation of clientRegistry and
removes two TODO comments.
2026-03-07 20:17:00 +08:00
Oleksandr Redko
bd200b1a3b fix: typos in comments, tests, functions (#5216) 2026-03-07 18:43:04 +08:00
fatedier
c70ceff370 fix: three high-severity bugs across nathole, proxy, and udp modules (#5214)
- pkg/nathole: add RLock when reading clientCfgs map in PreCheck path
  to prevent concurrent map read/write crash
- server/proxy: fix error variable shadowing in GetWorkConnFromPool
  that could return a closed connection with nil error
- pkg/util/net: check ListenUDP error before spawning goroutines
  and assign readConn to struct field so Close() works correctly
2026-03-07 13:36:02 +08:00
fatedier
bb3d0e7140 deduplicate common logic across proxy, visitor, and metrics modules (#5213)
- Replace duplicate parseBasicAuth with existing httppkg.ParseBasicAuth
- Extract buildDomains helper in BaseProxy for HTTP/HTTPS/TCPMux proxies
- Extract toProxyStats helper to deduplicate ProxyStats construction
- Extract startVisitorListener helper in BaseProxy for STCP/SUDP proxies
- Extract acceptLoop helper in BaseVisitor for STCP/XTCP visitors
2026-03-07 12:00:27 +08:00
fatedier
cf396563f8 client/proxy: unify work conn wrapping across all proxy types (#5212)
* client/proxy: extract wrapWorkConn to deduplicate UDP/SUDP connection wrapping

Move the repeated rate-limiting, encryption, and compression wrapping
logic from UDPProxy and SUDPProxy into a shared BaseProxy.wrapWorkConn
method, reducing ~18 lines of duplication in each proxy type.

* client/proxy: unify work conn wrapping with pooled compression for all proxy types

Refactor wrapWorkConn to accept encKey parameter and return
(io.ReadWriteCloser, recycleFn, error), enabling HandleTCPWorkConnection
to reuse the same limiter/encryption/compression pipeline.

Switch all proxy types from WithCompression to WithCompressionFromPool.
TCP non-plugin path calls recycleFn via defer after Join; plugin and
UDP/SUDP paths skip recycle (objects are GC'd safely, per golib contract).
2026-03-07 01:33:37 +08:00
fatedier
0b4f83cd04 pkg/config: use modern Go stdlib for sorting and string operations (#5210)
- slices.SortedFunc + maps.Values + cmp.Compare instead of manual
  map-to-slice collection + sort.Slice (source/aggregator.go)
- strings.CutSuffix instead of HasSuffix+TrimSuffix, and deduplicate
  error handling in BandwidthQuantity.UnmarshalString (types/types.go)
2026-03-06 23:13:29 +08:00
fatedier
e9f7a1a9f2 pkg: use modern Go stdlib functions to simplify code (#5209)
- strings.CutPrefix instead of HasPrefix+TrimPrefix (naming, legacy)
- slices.Contains instead of manual loop (plugin/server)
- min/max builtins instead of manual comparisons (nathole)
2026-03-06 22:14:46 +08:00
fatedier
d644593342 server/proxy: simplify HTTPS and TCPMux proxy domain registration (#5208)
Consolidate the separate custom-domain loop and subdomain block into a
single unified loop, matching the pattern already applied to HTTPProxy
in PR #5207. No behavioral change.
2026-03-06 21:31:29 +08:00
fatedier
427c4ca3ae server/proxy: simplify HTTP proxy domain registration by removing duplicate loop (#5207)
The Run() method had two nearly identical loop blocks for registering
custom domains and subdomain, with the same group/non-group registration
logic copy-pasted (~30 lines of duplication).

Consolidate by collecting all domains into a single slice first, then
iterating once with the shared registration logic. Also fixes a minor
inconsistency where the custom domain block used routeConfig.Domain in
CanonicalAddr but the subdomain block used tmpRouteConfig.Domain.
2026-03-06 21:17:30 +08:00
fatedier
f2d1f3739a pkg/util/xlog: fix AddPrefix not updating existing prefix due to range value copy (#5206)
In AddPrefix, the loop `for _, p := range l.prefixes` creates a copy
of each element. Assignments to p.Value and p.Priority only modify
the local copy, not the original slice element, causing updates to
existing prefixes to be silently lost.

This affects client/service.go where AddPrefix is called with
Name:"runID" on reconnection — the old runID value would persist
in log output instead of being updated to the new one.

Fix by using index-based access `l.prefixes[i]` to modify the
original slice element, and add break since prefix names are unique.
2026-03-06 20:44:40 +08:00
60 changed files with 576 additions and 772 deletions

View File

@@ -18,6 +18,7 @@ linters:
- lll - lll
- makezero - makezero
- misspell - misspell
- modernize
- prealloc - prealloc
- predeclared - predeclared
- revive - revive
@@ -47,6 +48,9 @@ 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,6 +16,7 @@ package proxy
import ( import (
"context" "context"
"fmt"
"io" "io"
"net" "net"
"reflect" "reflect"
@@ -122,6 +123,33 @@ func (pxy *BaseProxy) Close() {
} }
} }
// wrapWorkConn applies rate limiting, encryption, and compression
// to a work connection based on the proxy's transport configuration.
// The returned recycle function should be called when the stream is no longer in use
// to return compression resources to the pool. It is safe to not call recycle,
// in which case resources will be garbage collected normally.
func (pxy *BaseProxy) wrapWorkConn(conn net.Conn, encKey []byte) (io.ReadWriteCloser, func(), error) {
var rwc io.ReadWriteCloser = conn
if pxy.limiter != nil {
rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error {
return conn.Close()
})
}
if pxy.baseCfg.Transport.UseEncryption {
var err error
rwc, err = libio.WithEncryption(rwc, encKey)
if err != nil {
conn.Close()
return nil, nil, fmt.Errorf("create encryption stream error: %w", err)
}
}
var recycleFn func()
if pxy.baseCfg.Transport.UseCompression {
rwc, recycleFn = libio.WithCompressionFromPool(rwc)
}
return rwc, recycleFn, nil
}
func (pxy *BaseProxy) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) { func (pxy *BaseProxy) SetInWorkConnCallback(cb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool) {
pxy.inWorkConnCallback = cb pxy.inWorkConnCallback = cb
} }
@@ -139,30 +167,14 @@ 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, err = libio.WithEncryption(remote, encKey) remote, recycleFn, err := pxy.wrapWorkConn(workConn, encKey)
if err != nil { if err != nil {
workConn.Close() xl.Errorf("wrap work connection: %v", err)
xl.Errorf("create encryption stream error: %v", err) return
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
@@ -178,7 +190,6 @@ 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
} }
@@ -187,12 +198,18 @@ 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),
@@ -220,7 +237,4 @@ 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,7 +17,6 @@
package proxy package proxy
import ( import (
"io"
"net" "net"
"reflect" "reflect"
"strconv" "strconv"
@@ -25,17 +24,15 @@ 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.TypeOf(&v1.SUDPProxyConfig{}), NewSUDPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.SUDPProxyConfig](), NewSUDPProxy)
} }
type SUDPProxy struct { type SUDPProxy struct {
@@ -83,27 +80,13 @@ 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())
var rwc io.ReadWriteCloser = conn remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
var err error if err != nil {
if pxy.limiter != nil { xl.Errorf("wrap work connection: %v", err)
rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error { return
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 := conn workConn := netpkg.WrapReadWriteCloserToConn(remote, 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,24 +17,21 @@
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.TypeOf(&v1.UDPProxyConfig{}), NewUDPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.UDPProxyConfig](), NewUDPProxy)
} }
type UDPProxy struct { type UDPProxy struct {
@@ -94,28 +91,14 @@ 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()
var rwc io.ReadWriteCloser = conn remote, _, err := pxy.wrapWorkConn(conn, pxy.encryptionKey)
var err error if err != nil {
if pxy.limiter != nil { xl.Errorf("wrap work connection: %v", err)
rwc = libio.WrapReadWriteCloser(limit.NewReader(conn, pxy.limiter), limit.NewWriter(conn, pxy.limiter), func() error { return
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 = conn pxy.workConn = netpkg.WrapReadWriteCloserToConn(remote, 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.TypeOf(&v1.XTCPProxyConfig{}), NewXTCPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.XTCPProxyConfig](), NewXTCPProxy)
} }
type XTCPProxy struct { type XTCPProxy struct {

View File

@@ -15,18 +15,12 @@
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"
) )
@@ -42,10 +36,10 @@ func (sv *STCPVisitor) Run() (err error) {
if err != nil { if err != nil {
return return
} }
go sv.worker() go sv.acceptLoop(sv.l, "stcp local", sv.handleConn)
} }
go sv.internalConnWorker() go sv.acceptLoop(sv.internalLn, "stcp internal", sv.handleConn)
if sv.plugin != nil { if sv.plugin != nil {
sv.plugin.Start() sv.plugin.Start()
@@ -57,35 +51,10 @@ 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)
@@ -96,62 +65,21 @@ 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.helper.ConnectServer() visitorConn, err := sv.dialRawVisitorConn(sv.cfg.GetBaseConfig())
if err != nil { if err != nil {
xl.Warnf("dialRawVisitorConn error: %v", err)
tunnelErr = err tunnelErr = err
return return
} }
defer visitorConn.Close() defer visitorConn.Close()
now := time.Now().Unix() remote, recycleFn, err := wrapVisitorConn(visitorConn, sv.cfg.GetBaseConfig())
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("send newVisitorConnMsg to server error: %v", err) xl.Warnf("wrapVisitorConn 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,21 +16,17 @@ 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"
) )
@@ -76,6 +72,7 @@ func (sv *SUDPVisitor) dispatcher() {
var ( var (
visitorConn net.Conn visitorConn net.Conn
recycleFn func()
err error err error
firstPacket *msg.UDPPacket firstPacket *msg.UDPPacket
@@ -93,14 +90,17 @@ func (sv *SUDPVisitor) dispatcher() {
return return
} }
visitorConn, err = sv.getNewVisitorConn() visitorConn, recycleFn, 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.
sv.worker(visitorConn, firstPacket) func() {
defer recycleFn()
sv.worker(visitorConn, firstPacket)
}()
select { select {
case <-sv.checkCloseCh: case <-sv.checkCloseCh:
@@ -198,57 +198,17 @@ 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, error) { func (sv *SUDPVisitor) getNewVisitorConn() (net.Conn, func(), error) {
xl := xlog.FromContextSafe(sv.ctx) rawConn, err := sv.dialRawVisitorConn(sv.cfg.GetBaseConfig())
visitorConn, err := sv.helper.ConnectServer()
if err != nil { if err != nil {
return nil, fmt.Errorf("frpc connect frps error: %v", err) return nil, func() {}, 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 {
visitorConn.Close() 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 {
visitorConn.Close()
return nil, fmt.Errorf("frpc read newVisitorConnRespMsg error: %v", err)
}
_ = visitorConn.SetReadDeadline(time.Time{})
if newVisitorConnRespMsg.Error != "" {
visitorConn.Close()
return nil, fmt.Errorf("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
}
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)
visitorConn.Close()
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,13 +16,21 @@ 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"
) )
@@ -119,6 +127,18 @@ 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()
@@ -130,3 +150,57 @@ 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.worker() go sv.acceptLoop(sv.l, "xtcp local", sv.handleConn)
} }
go sv.internalConnWorker() go sv.acceptLoop(sv.internalLn, "xtcp internal", sv.handleConn)
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,30 +93,6 @@ 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 {
@@ -206,21 +182,14 @@ func (sv *XTCPVisitor) handleConn(userConn net.Conn) {
return return
} }
var muxConnRWCloser io.ReadWriteCloser = tunnelConn muxConnRWCloser, recycleFn, err := wrapVisitorConn(tunnelConn, sv.cfg.GetBaseConfig())
if sv.cfg.Transport.UseEncryption { if err != nil {
muxConnRWCloser, err = libio.WithEncryption(muxConnRWCloser, []byte(sv.cfg.SecretKey)) xl.Errorf("%v", err)
if err != nil { tunnelConn.Close()
xl.Errorf("create encryption stream error: %v", err) tunnelErr = err
tunnelConn.Close() 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")

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 pameters // ignore error here, because we can use command line parameters
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

@@ -171,15 +171,14 @@ 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 {
if !strings.HasPrefix(k, "plugin_header_") { k, ok := strings.CutPrefix(k, "plugin_header_")
if !ok || k == "" {
continue continue
} }
if k = strings.TrimPrefix(k, "plugin_header_"); k != "" { if out.Set == nil {
if out.Set == nil { out.Set = make(map[string]string)
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.TypeOf(TCPProxyConf{}), ProxyTypeTCP: reflect.TypeFor[TCPProxyConf](),
ProxyTypeUDP: reflect.TypeOf(UDPProxyConf{}), ProxyTypeUDP: reflect.TypeFor[UDPProxyConf](),
ProxyTypeTCPMUX: reflect.TypeOf(TCPMuxProxyConf{}), ProxyTypeTCPMUX: reflect.TypeFor[TCPMuxProxyConf](),
ProxyTypeHTTP: reflect.TypeOf(HTTPProxyConf{}), ProxyTypeHTTP: reflect.TypeFor[HTTPProxyConf](),
ProxyTypeHTTPS: reflect.TypeOf(HTTPSProxyConf{}), ProxyTypeHTTPS: reflect.TypeFor[HTTPSProxyConf](),
ProxyTypeSTCP: reflect.TypeOf(STCPProxyConf{}), ProxyTypeSTCP: reflect.TypeFor[STCPProxyConf](),
ProxyTypeXTCP: reflect.TypeOf(XTCPProxyConf{}), ProxyTypeXTCP: reflect.TypeFor[XTCPProxyConf](),
ProxyTypeSUDP: reflect.TypeOf(SUDPProxyConf{}), ProxyTypeSUDP: reflect.TypeFor[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 strings.HasPrefix(key, prefix) { if trimmed, ok := strings.CutPrefix(key, prefix); ok {
m[strings.TrimPrefix(key, prefix)] = value m[trimmed] = value
} }
} }

View File

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

View File

@@ -15,9 +15,11 @@
package source package source
import ( import (
"cmp"
"errors" "errors"
"fmt" "fmt"
"sort" "maps"
"slices"
"sync" "sync"
v1 "github.com/fatedier/frp/pkg/config/v1" v1 "github.com/fatedier/frp/pkg/config/v1"
@@ -97,21 +99,11 @@ 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 := make([]v1.ProxyConfigurer, 0, len(proxyMap)) proxies := slices.SortedFunc(maps.Values(proxyMap), func(x, y v1.ProxyConfigurer) int {
for _, p := range proxyMap { return cmp.Compare(x.GetBaseConfig().Name, y.GetBaseConfig().Name)
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 {
visitors := make([]v1.VisitorConfigurer, 0, len(visitorMap)) return cmp.Compare(x.GetBaseConfig().Name, y.GetBaseConfig().Name)
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,6 +196,27 @@ 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 := 0; i < len(firstRangeNumbers); i++ { for i := range firstRangeNumbers {
pairs = append(pairs, NumberPair{ pairs = append(pairs, NumberPair{
First: firstRangeNumbers[i], First: firstRangeNumbers[i],
Second: secondRangeNumbers[i], Second: secondRangeNumbers[i],

View File

@@ -70,24 +70,18 @@ func (q *BandwidthQuantity) UnmarshalString(s string) error {
f float64 f float64
err error err error
) )
switch { if fstr, ok := strings.CutSuffix(s, "MB"); ok {
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)
if err != nil { } else if fstr, ok := strings.CutSuffix(s, "KB"); ok {
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)
if err != nil { } else {
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))
@@ -143,8 +137,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.Split(str, ",") numRanges := strings.SplitSeq(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,6 +39,31 @@ 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.TypeOf(TCPProxyConfig{}), ProxyTypeTCP: reflect.TypeFor[TCPProxyConfig](),
ProxyTypeUDP: reflect.TypeOf(UDPProxyConfig{}), ProxyTypeUDP: reflect.TypeFor[UDPProxyConfig](),
ProxyTypeHTTP: reflect.TypeOf(HTTPProxyConfig{}), ProxyTypeHTTP: reflect.TypeFor[HTTPProxyConfig](),
ProxyTypeHTTPS: reflect.TypeOf(HTTPSProxyConfig{}), ProxyTypeHTTPS: reflect.TypeFor[HTTPSProxyConfig](),
ProxyTypeTCPMUX: reflect.TypeOf(TCPMuxProxyConfig{}), ProxyTypeTCPMUX: reflect.TypeFor[TCPMuxProxyConfig](),
ProxyTypeSTCP: reflect.TypeOf(STCPProxyConfig{}), ProxyTypeSTCP: reflect.TypeFor[STCPProxyConfig](),
ProxyTypeXTCP: reflect.TypeOf(XTCPProxyConfig{}), ProxyTypeXTCP: reflect.TypeFor[XTCPProxyConfig](),
ProxyTypeSUDP: reflect.TypeOf(SUDPProxyConfig{}), ProxyTypeSUDP: reflect.TypeFor[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.TypeOf(HTTP2HTTPSPluginOptions{}), PluginHTTP2HTTPS: reflect.TypeFor[HTTP2HTTPSPluginOptions](),
PluginHTTPProxy: reflect.TypeOf(HTTPProxyPluginOptions{}), PluginHTTPProxy: reflect.TypeFor[HTTPProxyPluginOptions](),
PluginHTTPS2HTTP: reflect.TypeOf(HTTPS2HTTPPluginOptions{}), PluginHTTPS2HTTP: reflect.TypeFor[HTTPS2HTTPPluginOptions](),
PluginHTTPS2HTTPS: reflect.TypeOf(HTTPS2HTTPSPluginOptions{}), PluginHTTPS2HTTPS: reflect.TypeFor[HTTPS2HTTPSPluginOptions](),
PluginHTTP2HTTP: reflect.TypeOf(HTTP2HTTPPluginOptions{}), PluginHTTP2HTTP: reflect.TypeFor[HTTP2HTTPPluginOptions](),
PluginSocks5: reflect.TypeOf(Socks5PluginOptions{}), PluginSocks5: reflect.TypeFor[Socks5PluginOptions](),
PluginStaticFile: reflect.TypeOf(StaticFilePluginOptions{}), PluginStaticFile: reflect.TypeFor[StaticFilePluginOptions](),
PluginUnixDomainSocket: reflect.TypeOf(UnixDomainSocketPluginOptions{}), PluginUnixDomainSocket: reflect.TypeFor[UnixDomainSocketPluginOptions](),
PluginTLS2Raw: reflect.TypeOf(TLS2RawPluginOptions{}), PluginTLS2Raw: reflect.TypeFor[TLS2RawPluginOptions](),
PluginVirtualNet: reflect.TypeOf(VirtualNetPluginOptions{}), PluginVirtualNet: reflect.TypeFor[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.TypeOf(STCPVisitorConfig{}), VisitorTypeSTCP: reflect.TypeFor[STCPVisitorConfig](),
VisitorTypeXTCP: reflect.TypeOf(XTCPVisitorConfig{}), VisitorTypeXTCP: reflect.TypeFor[XTCPVisitorConfig](),
VisitorTypeSUDP: reflect.TypeOf(SUDPVisitorConfig{}), VisitorTypeSUDP: reflect.TypeFor[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.TypeOf(VirtualNetVisitorPluginOptions{}), VisitorPluginVirtualNet: reflect.TypeFor[VirtualNetVisitorPluginOptions](),
} }
type VisitorPluginOptions interface { type VisitorPluginOptions interface {

View File

@@ -143,7 +143,6 @@ 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
} }
} }
@@ -155,7 +154,6 @@ 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
} }
} }
@@ -168,7 +166,6 @@ 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
} }
} }
@@ -181,7 +178,6 @@ 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
} }
} }
@@ -203,6 +199,25 @@ 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()
@@ -212,23 +227,7 @@ 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
} }
@@ -237,31 +236,9 @@ func (m *serverMetrics) GetProxiesByTypeAndName(proxyType string, proxyName stri
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
for name, proxyStats := range m.info.ProxyStatistics { proxyStats, ok := m.info.ProxyStatistics[proxyName]
if proxyStats.ProxyType != proxyType { if ok && proxyStats.ProxyType == proxyType {
continue res = toProxyStats(proxyName, proxyStats)
}
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
} }
@@ -272,21 +249,7 @@ 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 = &ProxyStats{ res = toProxyStats(proxyName, 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.TypeOf(&NatHoleResp{}).Elem().Name() var TypeNameNatHoleResp = reflect.TypeFor[NatHoleResp]().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,9 +16,8 @@ func StripUserPrefix(user, name string) string {
if user == "" { if user == "" {
return name return name
} }
prefix := user + "." if trimmed, ok := strings.CutPrefix(name, user+"."); ok {
if strings.HasPrefix(name, prefix) { return trimmed
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 := 0; i < len(behaviors); i++ { for i := range behaviors {
score := receiverScore score := receiverScore
if behaviors[i].A.Role == DetectRoleSender { if behaviors[i].A.Role == DetectRoleSender {
score = senderScore score = senderScore

View File

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

View File

@@ -152,7 +152,9 @@ 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

@@ -410,7 +410,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 i := 0; i < 10; i++ { for range 10 {
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 +420,7 @@ func sendSidMessageToRandomPorts(
return 0 return 0
} }
for i := 0; i < count; i++ { for range count {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return

View File

@@ -24,6 +24,7 @@ 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"
@@ -64,12 +65,7 @@ func (p *httpPlugin) Name() string {
} }
func (p *httpPlugin) IsSupport(op string) bool { func (p *httpPlugin) IsSupport(op string) bool {
for _, v := range p.options.Ops { return slices.Contains(p.options.Ops, op)
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,10 +153,7 @@ 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 = baseDelay * time.Duration(1<<uint(p.consecutiveErrors-1)) reconnectDelay = min(baseDelay*time.Duration(1<<uint(p.consecutiveErrors-1)), 300*time.Second)
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,6 +16,7 @@ package featuregate
import ( import (
"fmt" "fmt"
"maps"
"sort" "sort"
"strings" "strings"
"sync" "sync"
@@ -93,9 +94,7 @@ 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 := map[Feature]FeatureSpec{} known := map[Feature]FeatureSpec{}
for k, v := range defaultFeatures { maps.Copy(known, defaultFeatures)
known[k] = v
}
f := &featureGate{} f := &featureGate{}
f.known.Store(known) f.known.Store(known)
@@ -110,13 +109,9 @@ func (f *featureGate) SetFromMap(m map[string]bool) error {
// Copy existing state // Copy existing state
known := map[Feature]FeatureSpec{} known := map[Feature]FeatureSpec{}
for k, v := range f.known.Load().(map[Feature]FeatureSpec) { maps.Copy(known, f.known.Load().(map[Feature]FeatureSpec))
known[k] = v
}
enabled := map[Feature]bool{} enabled := map[Feature]bool{}
for k, v := range f.enabled.Load().(map[Feature]bool) { maps.Copy(enabled, 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 {
@@ -148,9 +143,7 @@ func (f *featureGate) Add(features map[Feature]FeatureSpec) error {
// Copy existing state // Copy existing state
known := map[Feature]FeatureSpec{} known := map[Feature]FeatureSpec{}
for k, v := range f.known.Load().(map[Feature]FeatureSpec) { maps.Copy(known, 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

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

View File

@@ -86,11 +86,7 @@ func (c *FakeUDPConn) Read(b []byte) (n int, err error) {
c.lastActive = time.Now() c.lastActive = time.Now()
c.mu.Unlock() c.mu.Unlock()
if len(b) < len(content) { n = min(len(b), len(content))
n = len(b)
} else {
n = len(content)
}
copy(b, content) copy(b, content)
return n, nil return n, nil
} }
@@ -168,11 +164,15 @@ 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

@@ -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.Split(rangeStr, ",") numRanges := strings.SplitSeq(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,31 +266,13 @@ 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, _, _ = parseBasicAuth(proxyAuth) user, _, _ = httppkg.ParseBasicAuth(proxyAuth)
} }
} }
if user == "" { if user == "" {

View File

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

View File

@@ -95,20 +95,33 @@ func (cm *ControlManager) Close() error {
return nil return nil
} }
type Control struct { // SessionContext encapsulates the input parameters for creating a new Control.
type SessionContext struct {
// all resource managers and controllers // all resource managers and controllers
rc *controller.ResourceController RC *controller.ResourceController
// proxy manager // proxy manager
pxyManager *proxy.Manager PxyManager *proxy.Manager
// plugin manager // plugin manager
pluginManager *plugin.Manager PluginManager *plugin.Manager
// verifies authentication based on selected method // verifies authentication based on selected method
authVerifier auth.Verifier AuthVerifier auth.Verifier
// key used for connection encryption // key used for connection encryption
encryptionKey []byte EncryptionKey []byte
// control connection
Conn net.Conn
// indicates whether the connection is encrypted
ConnEncrypted bool
// login message
LoginMsg *msg.Login
// server configuration
ServerCfg *v1.ServerConfig
// client registry
ClientRegistry *registry.ClientRegistry
}
type Control struct {
// session context
sessionCtx *SessionContext
// other components can use this to communicate with client // other components can use this to communicate with client
msgTransporter transport.MessageTransporter msgTransporter transport.MessageTransporter
@@ -117,12 +130,6 @@ 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
@@ -145,61 +152,34 @@ 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{}
} }
// TODO(fatedier): Referencing the implementation of frpc, encapsulate the input parameters as SessionContext. func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, error) {
func NewControl( poolCount := min(sessionCtx.LoginMsg.PoolCount, int(sessionCtx.ServerCfg.Transport.MaxPoolCount))
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{
rc: rc, sessionCtx: sessionCtx,
pxyManager: pxyManager, workConnCh: make(chan net.Conn, poolCount+10),
pluginManager: pluginManager, proxies: make(map[string]proxy.Proxy),
authVerifier: authVerifier, poolCount: poolCount,
encryptionKey: encryptionKey, portsUsedNum: 0,
conn: ctlConn, runID: sessionCtx.LoginMsg.RunID,
loginMsg: loginMsg, xl: xlog.FromContextSafe(ctx),
workConnCh: make(chan net.Conn, poolCount+10), ctx: ctx,
proxies: make(map[string]proxy.Proxy), doneCh: make(chan struct{}),
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 ctlConnEncrypted { if sessionCtx.ConnEncrypted {
cryptoRW, err := netpkg.NewCryptoReadWriter(ctl.conn, ctl.encryptionKey) cryptoRW, err := netpkg.NewCryptoReadWriter(sessionCtx.Conn, sessionCtx.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(ctl.conn) ctl.msgDispatcher = msg.NewDispatcher(sessionCtx.Conn)
} }
ctl.registerMsgHandlers() ctl.registerMsgHandlers()
ctl.msgTransporter = transport.NewMessageTransporter(ctl.msgDispatcher) ctl.msgTransporter = transport.NewMessageTransporter(ctl.msgDispatcher)
@@ -213,7 +193,7 @@ func (ctl *Control) Start() {
RunID: ctl.runID, RunID: ctl.runID,
Error: "", Error: "",
} }
_ = msg.WriteMsg(ctl.conn, loginRespMsg) _ = msg.WriteMsg(ctl.sessionCtx.Conn, loginRespMsg)
go func() { go func() {
for i := 0; i < ctl.poolCount; i++ { for i := 0; i < ctl.poolCount; i++ {
@@ -225,7 +205,7 @@ func (ctl *Control) Start() {
} }
func (ctl *Control) Close() error { func (ctl *Control) Close() error {
ctl.conn.Close() ctl.sessionCtx.Conn.Close()
return nil return nil
} }
@@ -233,7 +213,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.conn.Close() ctl.sessionCtx.Conn.Close()
} }
func (ctl *Control) RegisterWorkConn(conn net.Conn) error { func (ctl *Control) RegisterWorkConn(conn net.Conn) error {
@@ -291,7 +271,7 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
return return
} }
case <-time.After(time.Duration(ctl.serverCfg.UserConnTimeout) * time.Second): case <-time.After(time.Duration(ctl.sessionCtx.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
@@ -304,15 +284,15 @@ func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
} }
func (ctl *Control) heartbeatWorker() { func (ctl *Control) heartbeatWorker() {
if ctl.serverCfg.Transport.HeartbeatTimeout <= 0 { if ctl.sessionCtx.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.serverCfg.Transport.HeartbeatTimeout)*time.Second { if time.Since(ctl.lastPing.Load().(time.Time)) > time.Duration(ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout)*time.Second {
xl.Warnf("heartbeat timeout") xl.Warnf("heartbeat timeout")
ctl.conn.Close() ctl.sessionCtx.Conn.Close()
return return
} }
}, time.Second, ctl.doneCh) }, time.Second, ctl.doneCh)
@@ -323,6 +303,30 @@ 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
@@ -330,38 +334,23 @@ func (ctl *Control) worker() {
go ctl.msgDispatcher.Run() go ctl.msgDispatcher.Run()
<-ctl.msgDispatcher.Done() <-ctl.msgDispatcher.Done()
ctl.conn.Close() ctl.sessionCtx.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 ctl.proxies { for _, pxy := range proxies {
pxy.Close() ctl.closeProxy(pxy)
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.clientRegistry.MarkOfflineByRunID(ctl.runID) ctl.sessionCtx.ClientRegistry.MarkOfflineByRunID(ctl.runID)
xl.Infof("client exit success") xl.Infof("client exit success")
close(ctl.doneCh) close(ctl.doneCh)
} }
@@ -380,15 +369,11 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
inMsg := m.(*msg.NewProxy) inMsg := m.(*msg.NewProxy)
content := &plugin.NewProxyContent{ content := &plugin.NewProxyContent{
User: plugin.UserInfo{ User: ctl.loginUserInfo(),
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.pluginManager.NewProxy(content) retContent, err := ctl.sessionCtx.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)
@@ -401,15 +386,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.serverCfg.DetailedErrorsToClient)) err, lo.FromPtr(ctl.sessionCtx.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.loginMsg.ClientID clientID := ctl.sessionCtx.LoginMsg.ClientID
if clientID == "" { if clientID == "" {
clientID = ctl.loginMsg.RunID clientID = ctl.sessionCtx.LoginMsg.RunID
} }
metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.loginMsg.User, clientID) metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
} }
_ = ctl.msgDispatcher.Send(resp) _ = ctl.msgDispatcher.Send(resp)
} }
@@ -419,22 +404,18 @@ func (ctl *Control) handlePing(m msg.Message) {
inMsg := m.(*msg.Ping) inMsg := m.(*msg.Ping)
content := &plugin.PingContent{ content := &plugin.PingContent{
User: plugin.UserInfo{ User: ctl.loginUserInfo(),
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
RunID: ctl.loginMsg.RunID,
},
Ping: *inMsg, Ping: *inMsg,
} }
retContent, err := ctl.pluginManager.Ping(content) retContent, err := ctl.sessionCtx.PluginManager.Ping(content)
if err == nil { if err == nil {
inMsg = &retContent.Ping inMsg = &retContent.Ping
err = ctl.authVerifier.VerifyPing(inMsg) err = ctl.sessionCtx.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.serverCfg.DetailedErrorsToClient)), Error: util.GenerateResponseErrorString("invalid ping", err, lo.FromPtr(ctl.sessionCtx.ServerCfg.DetailedErrorsToClient)),
}) })
return return
} }
@@ -445,17 +426,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.rc.NatHoleController.HandleVisitor(inMsg, ctl.msgTransporter, ctl.loginMsg.User) ctl.sessionCtx.RC.NatHoleController.HandleVisitor(inMsg, ctl.msgTransporter, ctl.sessionCtx.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.rc.NatHoleController.HandleClient(inMsg, ctl.msgTransporter) ctl.sessionCtx.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.rc.NatHoleController.HandleReport(inMsg) ctl.sessionCtx.RC.NatHoleController.HandleReport(inMsg)
} }
func (ctl *Control) handleCloseProxy(m msg.Message) { func (ctl *Control) handleCloseProxy(m msg.Message) {
@@ -468,15 +449,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.serverCfg) pxyConf, err = config.NewProxyConfigurerFromMsg(pxyMsg, ctl.sessionCtx.ServerCfg)
if err != nil { if err != nil {
return return
} }
// User info // User info
userInfo := plugin.UserInfo{ userInfo := plugin.UserInfo{
User: ctl.loginMsg.User, User: ctl.sessionCtx.LoginMsg.User,
Metas: ctl.loginMsg.Metas, Metas: ctl.sessionCtx.LoginMsg.Metas,
RunID: ctl.runID, RunID: ctl.runID,
} }
@@ -484,22 +465,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.loginMsg, LoginMsg: ctl.sessionCtx.LoginMsg,
PoolCount: ctl.poolCount, PoolCount: ctl.poolCount,
ResourceController: ctl.rc, ResourceController: ctl.sessionCtx.RC,
GetWorkConnFn: ctl.GetWorkConn, GetWorkConnFn: ctl.GetWorkConn,
Configurer: pxyConf, Configurer: pxyConf,
ServerCfg: ctl.serverCfg, ServerCfg: ctl.sessionCtx.ServerCfg,
EncryptionKey: ctl.encryptionKey, EncryptionKey: ctl.sessionCtx.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.serverCfg.MaxPortsPerClient > 0 { if ctl.sessionCtx.ServerCfg.MaxPortsPerClient > 0 {
ctl.mu.Lock() ctl.mu.Lock()
if ctl.portsUsedNum+pxy.GetUsedPortsNum() > int(ctl.serverCfg.MaxPortsPerClient) { if ctl.portsUsedNum+pxy.GetUsedPortsNum() > int(ctl.sessionCtx.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
@@ -516,7 +497,7 @@ func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err
}() }()
} }
if ctl.pxyManager.Exist(pxyMsg.ProxyName) { if ctl.sessionCtx.PxyManager.Exist(pxyMsg.ProxyName) {
err = fmt.Errorf("proxy [%s] already exists", pxyMsg.ProxyName) err = fmt.Errorf("proxy [%s] already exists", pxyMsg.ProxyName)
return return
} }
@@ -531,7 +512,7 @@ func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err
} }
}() }()
err = ctl.pxyManager.Add(pxyMsg.ProxyName, pxy) err = ctl.sessionCtx.PxyManager.Add(pxyMsg.ProxyName, pxy)
if err != nil { if err != nil {
return return
} }
@@ -550,28 +531,12 @@ func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
return return
} }
if ctl.serverCfg.MaxPortsPerClient > 0 { if ctl.sessionCtx.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()
metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type) ctl.closeProxy(pxy)
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

@@ -31,7 +31,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeOf(&v1.HTTPProxyConfig{}), NewHTTPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.HTTPProxyConfig](), NewHTTPProxy)
} }
type HTTPProxy struct { type HTTPProxy struct {
@@ -75,16 +75,13 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
} }
}() }()
addrs := make([]string, 0) domains := pxy.buildDomains(pxy.cfg.CustomDomains, pxy.cfg.SubDomain)
for _, domain := range pxy.cfg.CustomDomains {
if domain == "" {
continue
}
addrs := make([]string, 0)
for _, domain := range domains {
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
@@ -93,12 +90,10 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
if err != nil { if err != nil {
return 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)
}) })
} else { } else {
// no group
err = pxy.rc.HTTPReverseProxy.Register(routeConfig) err = pxy.rc.HTTPReverseProxy.Register(routeConfig)
if err != nil { if err != nil {
return return
@@ -112,39 +107,6 @@ func (pxy *HTTPProxy) Run() (remoteAddr string, err error) {
routeConfig.Domain, routeConfig.Location, pxy.cfg.LoadBalancer.Group, pxy.cfg.RouteByHTTPUser) routeConfig.Domain, routeConfig.Location, pxy.cfg.LoadBalancer.Group, pxy.cfg.RouteByHTTPUser)
} }
} }
if pxy.cfg.SubDomain != "" {
routeConfig.Domain = pxy.cfg.SubDomain + "." + pxy.serverCfg.SubDomainHost
for _, location := range locations {
routeConfig.Location = location
tmpRouteConfig := routeConfig
// handle group
if pxy.cfg.LoadBalancer.Group != "" {
err = pxy.rc.HTTPGroupCtl.Register(pxy.name, pxy.cfg.LoadBalancer.Group, pxy.cfg.LoadBalancer.GroupKey, routeConfig)
if err != nil {
return
}
pxy.closeFuncs = append(pxy.closeFuncs, func() {
pxy.rc.HTTPGroupCtl.UnRegister(pxy.name, pxy.cfg.LoadBalancer.Group, tmpRouteConfig)
})
} else {
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(tmpRouteConfig.Domain, pxy.serverCfg.VhostHTTPPort))
xl.Infof("http proxy listen for host [%s] location [%s] group [%s], routeByHTTPUser [%s]",
routeConfig.Domain, routeConfig.Location, pxy.cfg.LoadBalancer.Group, pxy.cfg.RouteByHTTPUser)
}
}
remoteAddr = strings.Join(addrs, ",") remoteAddr = strings.Join(addrs, ",")
return return
} }

View File

@@ -25,7 +25,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeOf(&v1.HTTPSProxyConfig{}), NewHTTPSProxy) RegisterProxyFactory(reflect.TypeFor[*v1.HTTPSProxyConfig](), NewHTTPSProxy)
} }
type HTTPSProxy struct { type HTTPSProxy struct {
@@ -53,23 +53,10 @@ 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 pxy.cfg.CustomDomains { for _, domain := range domains {
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,6 +161,7 @@ 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
} }
@@ -173,6 +174,36 @@ 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.TypeOf(&v1.STCPProxyConfig{}), NewSTCPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.STCPProxyConfig](), NewSTCPProxy)
} }
type STCPProxy struct { type STCPProxy struct {
@@ -41,21 +41,7 @@ func NewSTCPProxy(baseProxy *BaseProxy) Proxy {
} }
func (pxy *STCPProxy) Run() (remoteAddr string, err error) { func (pxy *STCPProxy) Run() (remoteAddr string, err error) {
xl := pxy.xl err = pxy.startVisitorListener(pxy.cfg.Secretkey, pxy.cfg.AllowUsers, "stcp")
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.TypeOf(&v1.SUDPProxyConfig{}), NewSUDPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.SUDPProxyConfig](), NewSUDPProxy)
} }
type SUDPProxy struct { type SUDPProxy struct {
@@ -41,21 +41,7 @@ func NewSUDPProxy(baseProxy *BaseProxy) Proxy {
} }
func (pxy *SUDPProxy) Run() (remoteAddr string, err error) { func (pxy *SUDPProxy) Run() (remoteAddr string, err error) {
xl := pxy.xl err = pxy.startVisitorListener(pxy.cfg.Secretkey, pxy.cfg.AllowUsers, "sudp")
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.TypeOf(&v1.TCPProxyConfig{}), NewTCPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.TCPProxyConfig](), NewTCPProxy)
} }
type TCPProxy struct { type TCPProxy struct {

View File

@@ -26,7 +26,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeOf(&v1.TCPMuxProxyConfig{}), NewTCPMuxProxy) RegisterProxyFactory(reflect.TypeFor[*v1.TCPMuxProxyConfig](), NewTCPMuxProxy)
} }
type TCPMuxProxy struct { type TCPMuxProxy struct {
@@ -72,26 +72,16 @@ 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 pxy.cfg.CustomDomains { for _, domain := range domains {
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.TypeOf(&v1.UDPProxyConfig{}), NewUDPProxy) RegisterProxyFactory(reflect.TypeFor[*v1.UDPProxyConfig](), NewUDPProxy)
} }
type UDPProxy struct { type UDPProxy struct {

View File

@@ -24,7 +24,7 @@ import (
) )
func init() { func init() {
RegisterProxyFactory(reflect.TypeOf(&v1.XTCPProxyConfig{}), NewXTCPProxy) RegisterProxyFactory(reflect.TypeFor[*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, passthough: %v", address, cfg.TCPMuxPassthrough) log.Infof("tcpmux httpconnect multiplexer listen on %s, passthrough: %v", address, cfg.TCPMuxPassthrough)
} }
// Init all plugins // Init all plugins
@@ -604,8 +604,18 @@ func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, inter
return err return err
} }
// TODO(fatedier): use SessionContext ctl, err := NewControl(ctx, &SessionContext{
ctl, err := NewControl(ctx, svr.rc, svr.pxyManager, svr.pluginManager, authVerifier, svr.auth.EncryptionKey(), ctlConn, !internal, loginMsg, svr.cfg) RC: svr.rc,
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
@@ -626,7 +636,6 @@ 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()
@@ -652,9 +661,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.loginMsg.User, User: ctl.sessionCtx.LoginMsg.User,
Metas: ctl.loginMsg.Metas, Metas: ctl.sessionCtx.LoginMsg.Metas,
RunID: ctl.loginMsg.RunID, RunID: ctl.sessionCtx.LoginMsg.RunID,
}, },
NewWorkConn: *newMsg, NewWorkConn: *newMsg,
} }
@@ -662,7 +671,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.authVerifier.VerifyNewWorkConn(newMsg) err = ctl.sessionCtx.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)
@@ -683,7 +692,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.loginMsg.User visitorUser = ctl.sessionCtx.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,6 +2,7 @@ package framework
import ( import (
"fmt" "fmt"
"maps"
"os" "os"
"path/filepath" "path/filepath"
"time" "time"
@@ -20,9 +21,7 @@ func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []str
ExpectNoError(err) ExpectNoError(err)
ExpectTrue(len(templates) > 0) ExpectTrue(len(templates) > 0)
for name, port := range ports { maps.Copy(f.usedPorts, 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,7 +26,8 @@ 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
clientConf := consts.LegacyDefaultClientConfig var clientConf strings.Builder
clientConf.WriteString(consts.LegacyDefaultClientConfig)
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -78,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 += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
@@ -102,7 +103,8 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
vhost_http_port = %d vhost_http_port = %d
`, vhostHTTPPort) `, vhostHTTPPort)
clientConf := consts.LegacyDefaultClientConfig var clientConf strings.Builder
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(`
@@ -147,13 +149,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 += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
for _, test := range tests { for _, test := range tests {
for _, domain := range strings.Split(test.customDomains, ",") { for domain := range strings.SplitSeq(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain(test.proxyName + "-" + domain). Explain(test.proxyName + "-" + domain).
@@ -185,7 +187,8 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
`, vhostHTTPSPort) `, vhostHTTPSPort)
localPort := f.AllocPort() localPort := f.AllocPort()
clientConf := consts.LegacyDefaultClientConfig var clientConf strings.Builder
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]
@@ -229,10 +232,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 += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
tlsConfig, err := transport.NewServerTLSConfig("", "", "") tlsConfig, err := transport.NewServerTLSConfig("", "", "")
framework.ExpectNoError(err) framework.ExpectNoError(err)
@@ -244,7 +247,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.Split(test.customDomains, ",") { for domain := range strings.SplitSeq(test.customDomains, ",") {
domain = strings.TrimSpace(domain) domain = strings.TrimSpace(domain)
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain(test.proxyName + "-" + domain). Explain(test.proxyName + "-" + domain).
@@ -282,9 +285,12 @@ 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
clientServerConf := consts.LegacyDefaultClientConfig + "\nuser = user1" var clientServerConf strings.Builder
clientVisitorConf := consts.LegacyDefaultClientConfig + "\nuser = user1" clientServerConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user1")
clientUser2VisitorConf := consts.LegacyDefaultClientConfig + "\nuser = user2" var clientVisitorConf strings.Builder
clientVisitorConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user1")
var clientUser2VisitorConf strings.Builder
clientUser2VisitorConf.WriteString(consts.LegacyDefaultClientConfig + "\nuser = user2")
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -400,20 +406,20 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientServerConf += getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n" clientServerConf.WriteString(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 += config clientUser2VisitorConf.WriteString(config)
} else { } else {
clientVisitorConf += config clientVisitorConf.WriteString(config)
} }
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientServerConf, clientVisitorConf, clientUser2VisitorConf}) f.RunProcesses([]string{serverConf}, []string{clientServerConf.String(), clientVisitorConf.String(), clientUser2VisitorConf.String()})
for _, test := range tests { for _, test := range tests {
timeout := time.Second timeout := time.Second
@@ -440,7 +446,8 @@ 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
clientConf := consts.LegacyDefaultClientConfig var clientConf strings.Builder
clientConf.WriteString(consts.LegacyDefaultClientConfig)
tcpmuxHTTPConnectPortName := port.GenName("TCPMUX") tcpmuxHTTPConnectPortName := port.GenName("TCPMUX")
serverConf += fmt.Sprintf(` serverConf += fmt.Sprintf(`
@@ -483,14 +490,14 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf += getProxyConf(test.proxyName, test.extraConfig) + "\n" clientConf.WriteString(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}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
// 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-allowded-in-range] [tcp-allowed-in-range]
type = tcp type = tcp
local_port = {{ .%s }} local_port = {{ .%s }}
remote_port = {{ .%s }} remote_port = {{ .%s }}

View File

@@ -48,12 +48,10 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
return true return true
}) })
} }
for i := 0; i < 10; i++ { for range 10 {
wait.Add(1) wait.Go(func() {
go func() {
defer wait.Done()
expectFn() expectFn()
}() })
} }
wait.Wait() wait.Wait()
@@ -94,7 +92,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := 0; i < 10; i++ { for i := range 10 {
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":
@@ -150,7 +148,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// check foo and bar is ok // check foo and bar is ok
results := []string{} results := []string{}
for i := 0; i < 10; i++ { for range 10 {
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
@@ -161,7 +159,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 i := 0; i < 10; i++ { for range 10 {
framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
} }
@@ -169,7 +167,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 i := 0; i < 10; i++ { for range 10 {
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,6 +4,7 @@ import (
"crypto/tls" "crypto/tls"
"fmt" "fmt"
"strconv" "strconv"
"strings"
"github.com/onsi/ginkgo/v2" "github.com/onsi/ginkgo/v2"
@@ -22,7 +23,8 @@ 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
clientConf := consts.LegacyDefaultClientConfig var clientConf strings.Builder
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(`
@@ -65,10 +67,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 += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
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 i := 0; i < 20; i++ { for range 20 {
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.hanldeConn(conn) go c.handleConn(conn)
} }
} }
func (c *TunnelClient) hanldeConn(conn net.Conn) { func (c *TunnelClient) handleConn(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,7 +26,8 @@ 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
clientConf := consts.DefaultClientConfig var clientConf strings.Builder
clientConf.WriteString(consts.DefaultClientConfig)
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -79,10 +80,10 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
for _, test := range tests { for _, test := range tests {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
@@ -103,7 +104,8 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
vhostHTTPPort = %d vhostHTTPPort = %d
`, vhostHTTPPort) `, vhostHTTPPort)
clientConf := consts.DefaultClientConfig var clientConf strings.Builder
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(`
@@ -149,13 +151,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 += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
for _, test := range tests { for _, test := range tests {
for _, domain := range strings.Split(test.customDomains, ",") { for domain := range strings.SplitSeq(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, "]\"")
@@ -189,7 +191,8 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
`, vhostHTTPSPort) `, vhostHTTPSPort)
localPort := f.AllocPort() localPort := f.AllocPort()
clientConf := consts.DefaultClientConfig var clientConf strings.Builder
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]]
@@ -234,10 +237,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 += getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, tests[i].customDomains, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
tlsConfig, err := transport.NewServerTLSConfig("", "", "") tlsConfig, err := transport.NewServerTLSConfig("", "", "")
framework.ExpectNoError(err) framework.ExpectNoError(err)
@@ -249,7 +252,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.Split(test.customDomains, ",") { for domain := range strings.SplitSeq(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, "]\"")
@@ -289,9 +292,12 @@ 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
clientServerConf := consts.DefaultClientConfig + "\nuser = \"user1\"" var clientServerConf strings.Builder
clientVisitorConf := consts.DefaultClientConfig + "\nuser = \"user1\"" clientServerConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user1\"")
clientUser2VisitorConf := consts.DefaultClientConfig + "\nuser = \"user2\"" var clientVisitorConf strings.Builder
clientVisitorConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user1\"")
var clientUser2VisitorConf strings.Builder
clientUser2VisitorConf.WriteString(consts.DefaultClientConfig + "\nuser = \"user2\"")
localPortName := "" localPortName := ""
protocol := "tcp" protocol := "tcp"
@@ -407,20 +413,20 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientServerConf += getProxyServerConf(test.proxyName, test.commonExtraConfig+"\n"+test.proxyExtraConfig) + "\n" clientServerConf.WriteString(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 += config clientUser2VisitorConf.WriteString(config)
} else { } else {
clientVisitorConf += config clientVisitorConf.WriteString(config)
} }
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientServerConf, clientVisitorConf, clientUser2VisitorConf}) f.RunProcesses([]string{serverConf}, []string{clientServerConf.String(), clientVisitorConf.String(), clientUser2VisitorConf.String()})
for _, test := range tests { for _, test := range tests {
timeout := time.Second timeout := time.Second
@@ -447,7 +453,8 @@ 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
clientConf := consts.DefaultClientConfig var clientConf strings.Builder
clientConf.WriteString(consts.DefaultClientConfig)
tcpmuxHTTPConnectPortName := port.GenName("TCPMUX") tcpmuxHTTPConnectPortName := port.GenName("TCPMUX")
serverConf += fmt.Sprintf(` serverConf += fmt.Sprintf(`
@@ -491,14 +498,14 @@ var _ = ginkgo.Describe("[Feature: Basic]", func() {
// build all client config // build all client config
for _, test := range tests { for _, test := range tests {
clientConf += getProxyConf(test.proxyName, test.extraConfig) + "\n" clientConf.WriteString(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}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
// 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-allowded-in-range" name = "tcp-allowed-in-range"
type = "tcp" type = "tcp"
localPort = {{ .%s }} localPort = {{ .%s }}
remotePort = {{ .%s }} remotePort = {{ .%s }}

View File

@@ -50,12 +50,10 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
return true return true
}) })
} }
for i := 0; i < 10; i++ { for range 10 {
wait.Add(1) wait.Go(func() {
go func() {
defer wait.Done()
expectFn() expectFn()
}() })
} }
wait.Wait() wait.Wait()
@@ -98,7 +96,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := 0; i < 10; i++ { for i := range 10 {
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":
@@ -163,7 +161,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
fooCount := 0 fooCount := 0
barCount := 0 barCount := 0
for i := 0; i < 10; i++ { for i := range 10 {
framework.NewRequestExpect(f). framework.NewRequestExpect(f).
Explain("times " + strconv.Itoa(i)). Explain("times " + strconv.Itoa(i)).
Port(vhostHTTPSPort). Port(vhostHTTPSPort).
@@ -230,7 +228,7 @@ var _ = ginkgo.Describe("[Feature: Group]", func() {
// check foo and bar is ok // check foo and bar is ok
results := []string{} results := []string{}
for i := 0; i < 10; i++ { for range 10 {
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
@@ -241,7 +239,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 i := 0; i < 10; i++ { for range 10 {
framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure() framework.NewRequestExpect(f).Port(remotePort).ExpectResp([]byte("foo")).Ensure()
} }
@@ -249,7 +247,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 i := 0; i < 10; i++ { for range 10 {
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,6 +5,7 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"github.com/onsi/ginkgo/v2" "github.com/onsi/ginkgo/v2"
@@ -23,7 +24,8 @@ 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
clientConf := consts.DefaultClientConfig var clientConf strings.Builder
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(`
@@ -69,10 +71,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 += getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n" clientConf.WriteString(getProxyConf(test.proxyName, test.portName, test.extraConfig) + "\n")
} }
// run frps and frpc // run frps and frpc
f.RunProcesses([]string{serverConf}, []string{clientConf}) f.RunProcesses([]string{serverConf}, []string{clientConf.String()})
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()