|
@@ -16,32 +16,25 @@ package client
|
|
|
|
|
|
import (
|
|
|
"context"
|
|
|
- "crypto/tls"
|
|
|
+ "errors"
|
|
|
"fmt"
|
|
|
- "io"
|
|
|
"net"
|
|
|
"runtime"
|
|
|
- "strconv"
|
|
|
- "strings"
|
|
|
"sync"
|
|
|
- "sync/atomic"
|
|
|
"time"
|
|
|
|
|
|
"github.com/fatedier/golib/crypto"
|
|
|
- libdial "github.com/fatedier/golib/net/dial"
|
|
|
- fmux "github.com/hashicorp/yamux"
|
|
|
- quic "github.com/quic-go/quic-go"
|
|
|
"github.com/samber/lo"
|
|
|
|
|
|
- "github.com/fatedier/frp/assets"
|
|
|
+ "github.com/fatedier/frp/client/proxy"
|
|
|
"github.com/fatedier/frp/pkg/auth"
|
|
|
v1 "github.com/fatedier/frp/pkg/config/v1"
|
|
|
"github.com/fatedier/frp/pkg/msg"
|
|
|
- "github.com/fatedier/frp/pkg/transport"
|
|
|
+ httppkg "github.com/fatedier/frp/pkg/util/http"
|
|
|
"github.com/fatedier/frp/pkg/util/log"
|
|
|
- utilnet "github.com/fatedier/frp/pkg/util/net"
|
|
|
- "github.com/fatedier/frp/pkg/util/util"
|
|
|
+ netpkg "github.com/fatedier/frp/pkg/util/net"
|
|
|
"github.com/fatedier/frp/pkg/util/version"
|
|
|
+ "github.com/fatedier/frp/pkg/util/wait"
|
|
|
"github.com/fatedier/frp/pkg/util/xlog"
|
|
|
)
|
|
|
|
|
@@ -49,212 +42,197 @@ func init() {
|
|
|
crypto.DefaultSalt = "frp"
|
|
|
}
|
|
|
|
|
|
-// Service is a client service.
|
|
|
-type Service struct {
|
|
|
- // uniq id got from frps, attach it in loginMsg
|
|
|
- runID string
|
|
|
+type cancelErr struct {
|
|
|
+ Err error
|
|
|
+}
|
|
|
|
|
|
- // manager control connection with server
|
|
|
- ctl *Control
|
|
|
+func (e cancelErr) Error() string {
|
|
|
+ return e.Err.Error()
|
|
|
+}
|
|
|
+
|
|
|
+// ServiceOptions contains options for creating a new client service.
|
|
|
+type ServiceOptions struct {
|
|
|
+ Common *v1.ClientCommonConfig
|
|
|
+ ProxyCfgs []v1.ProxyConfigurer
|
|
|
+ VisitorCfgs []v1.VisitorConfigurer
|
|
|
+
|
|
|
+ // ConfigFilePath is the path to the configuration file used to initialize.
|
|
|
+ // If it is empty, it means that the configuration file is not used for initialization.
|
|
|
+ // It may be initialized using command line parameters or called directly.
|
|
|
+ ConfigFilePath string
|
|
|
+
|
|
|
+ // ClientSpec is the client specification that control the client behavior.
|
|
|
+ ClientSpec *msg.ClientSpec
|
|
|
+
|
|
|
+ // ConnectorCreator is a function that creates a new connector to make connections to the server.
|
|
|
+ // The Connector shields the underlying connection details, whether it is through TCP or QUIC connection,
|
|
|
+ // and regardless of whether multiplexing is used.
|
|
|
+ //
|
|
|
+ // If it is not set, the default frpc connector will be used.
|
|
|
+ // By using a custom Connector, it can be used to implement a VirtualClient, which connects to frps
|
|
|
+ // through a pipe instead of a real physical connection.
|
|
|
+ ConnectorCreator func(context.Context, *v1.ClientCommonConfig) Connector
|
|
|
+
|
|
|
+ // HandleWorkConnCb is a callback function that is called when a new work connection is created.
|
|
|
+ //
|
|
|
+ // If it is not set, the default frpc implementation will be used.
|
|
|
+ HandleWorkConnCb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool
|
|
|
+}
|
|
|
+
|
|
|
+// setServiceOptionsDefault sets the default values for ServiceOptions.
|
|
|
+func setServiceOptionsDefault(options *ServiceOptions) {
|
|
|
+ if options.Common != nil {
|
|
|
+ options.Common.Complete()
|
|
|
+ }
|
|
|
+ if options.ConnectorCreator == nil {
|
|
|
+ options.ConnectorCreator = NewConnector
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// Service is the client service that connects to frps and provides proxy services.
|
|
|
+type Service struct {
|
|
|
ctlMu sync.RWMutex
|
|
|
+ // manager control connection with server
|
|
|
+ ctl *Control
|
|
|
+ // Uniq id got from frps, it will be attached to loginMsg.
|
|
|
+ runID string
|
|
|
|
|
|
// Sets authentication based on selected method
|
|
|
authSetter auth.Setter
|
|
|
|
|
|
- cfg *v1.ClientCommonConfig
|
|
|
- pxyCfgs []v1.ProxyConfigurer
|
|
|
- visitorCfgs []v1.VisitorConfigurer
|
|
|
+ // web server for admin UI and apis
|
|
|
+ webServer *httppkg.Server
|
|
|
+
|
|
|
cfgMu sync.RWMutex
|
|
|
+ common *v1.ClientCommonConfig
|
|
|
+ proxyCfgs []v1.ProxyConfigurer
|
|
|
+ visitorCfgs []v1.VisitorConfigurer
|
|
|
+ clientSpec *msg.ClientSpec
|
|
|
|
|
|
// The configuration file used to initialize this client, or an empty
|
|
|
// string if no configuration file was used.
|
|
|
- cfgFile string
|
|
|
-
|
|
|
- exit uint32 // 0 means not exit
|
|
|
+ configFilePath string
|
|
|
|
|
|
// service context
|
|
|
ctx context.Context
|
|
|
// call cancel to stop service
|
|
|
- cancel context.CancelFunc
|
|
|
-}
|
|
|
+ cancel context.CancelCauseFunc
|
|
|
+ gracefulShutdownDuration time.Duration
|
|
|
|
|
|
-func NewService(
|
|
|
- cfg *v1.ClientCommonConfig,
|
|
|
- pxyCfgs []v1.ProxyConfigurer,
|
|
|
- visitorCfgs []v1.VisitorConfigurer,
|
|
|
- cfgFile string,
|
|
|
-) (svr *Service, err error) {
|
|
|
- svr = &Service{
|
|
|
- authSetter: auth.NewAuthSetter(cfg.Auth),
|
|
|
- cfg: cfg,
|
|
|
- cfgFile: cfgFile,
|
|
|
- pxyCfgs: pxyCfgs,
|
|
|
- visitorCfgs: visitorCfgs,
|
|
|
- ctx: context.Background(),
|
|
|
- exit: 0,
|
|
|
- }
|
|
|
- return
|
|
|
+ connectorCreator func(context.Context, *v1.ClientCommonConfig) Connector
|
|
|
+ handleWorkConnCb func(*v1.ProxyBaseConfig, net.Conn, *msg.StartWorkConn) bool
|
|
|
}
|
|
|
|
|
|
-func (svr *Service) GetController() *Control {
|
|
|
- svr.ctlMu.RLock()
|
|
|
- defer svr.ctlMu.RUnlock()
|
|
|
- return svr.ctl
|
|
|
+func NewService(options ServiceOptions) (*Service, error) {
|
|
|
+ setServiceOptionsDefault(&options)
|
|
|
+
|
|
|
+ var webServer *httppkg.Server
|
|
|
+ if options.Common.WebServer.Port > 0 {
|
|
|
+ ws, err := httppkg.NewServer(options.Common.WebServer)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ webServer = ws
|
|
|
+ }
|
|
|
+ s := &Service{
|
|
|
+ ctx: context.Background(),
|
|
|
+ authSetter: auth.NewAuthSetter(options.Common.Auth),
|
|
|
+ webServer: webServer,
|
|
|
+ common: options.Common,
|
|
|
+ configFilePath: options.ConfigFilePath,
|
|
|
+ proxyCfgs: options.ProxyCfgs,
|
|
|
+ visitorCfgs: options.VisitorCfgs,
|
|
|
+ clientSpec: options.ClientSpec,
|
|
|
+ connectorCreator: options.ConnectorCreator,
|
|
|
+ handleWorkConnCb: options.HandleWorkConnCb,
|
|
|
+ }
|
|
|
+ if webServer != nil {
|
|
|
+ webServer.RouteRegister(s.registerRouteHandlers)
|
|
|
+ }
|
|
|
+ return s, nil
|
|
|
}
|
|
|
|
|
|
func (svr *Service) Run(ctx context.Context) error {
|
|
|
- ctx, cancel := context.WithCancel(ctx)
|
|
|
- svr.ctx = xlog.NewContext(ctx, xlog.New())
|
|
|
+ ctx, cancel := context.WithCancelCause(ctx)
|
|
|
+ svr.ctx = xlog.NewContext(ctx, xlog.FromContextSafe(ctx))
|
|
|
svr.cancel = cancel
|
|
|
|
|
|
- xl := xlog.FromContextSafe(svr.ctx)
|
|
|
-
|
|
|
// set custom DNSServer
|
|
|
- if svr.cfg.DNSServer != "" {
|
|
|
- dnsAddr := svr.cfg.DNSServer
|
|
|
- if _, _, err := net.SplitHostPort(dnsAddr); err != nil {
|
|
|
- dnsAddr = net.JoinHostPort(dnsAddr, "53")
|
|
|
- }
|
|
|
- // Change default dns server for frpc
|
|
|
- net.DefaultResolver = &net.Resolver{
|
|
|
- PreferGo: true,
|
|
|
- Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
|
|
|
- return net.Dial("udp", dnsAddr)
|
|
|
- },
|
|
|
- }
|
|
|
+ if svr.common.DNSServer != "" {
|
|
|
+ netpkg.SetDefaultDNSAddress(svr.common.DNSServer)
|
|
|
}
|
|
|
|
|
|
- // login to frps
|
|
|
- for {
|
|
|
- conn, cm, err := svr.login()
|
|
|
- if err != nil {
|
|
|
- xl.Warn("login to server failed: %v", err)
|
|
|
-
|
|
|
- // if login_fail_exit is true, just exit this program
|
|
|
- // otherwise sleep a while and try again to connect to server
|
|
|
- if lo.FromPtr(svr.cfg.LoginFailExit) {
|
|
|
- return err
|
|
|
- }
|
|
|
- util.RandomSleep(5*time.Second, 0.9, 1.1)
|
|
|
- } else {
|
|
|
- // login success
|
|
|
- ctl := NewControl(svr.ctx, svr.runID, conn, cm, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.authSetter)
|
|
|
- ctl.Run()
|
|
|
- svr.ctlMu.Lock()
|
|
|
- svr.ctl = ctl
|
|
|
- svr.ctlMu.Unlock()
|
|
|
- break
|
|
|
- }
|
|
|
+ // first login to frps
|
|
|
+ svr.loopLoginUntilSuccess(10*time.Second, lo.FromPtr(svr.common.LoginFailExit))
|
|
|
+ if svr.ctl == nil {
|
|
|
+ cancelCause := cancelErr{}
|
|
|
+ _ = errors.As(context.Cause(svr.ctx), &cancelCause)
|
|
|
+ return fmt.Errorf("login to the server failed: %v. With loginFailExit enabled, no additional retries will be attempted", cancelCause.Err)
|
|
|
}
|
|
|
|
|
|
go svr.keepControllerWorking()
|
|
|
|
|
|
- if svr.cfg.WebServer.Port != 0 {
|
|
|
- // Init admin server assets
|
|
|
- assets.Load(svr.cfg.WebServer.AssetsDir)
|
|
|
-
|
|
|
- address := net.JoinHostPort(svr.cfg.WebServer.Addr, strconv.Itoa(svr.cfg.WebServer.Port))
|
|
|
- err := svr.RunAdminServer(address)
|
|
|
- if err != nil {
|
|
|
- log.Warn("run admin server error: %v", err)
|
|
|
- }
|
|
|
- log.Info("admin server listen on %s:%d", svr.cfg.WebServer.Addr, svr.cfg.WebServer.Port)
|
|
|
+ if svr.webServer != nil {
|
|
|
+ go func() {
|
|
|
+ log.Info("admin server listen on %s", svr.webServer.Address())
|
|
|
+ if err := svr.webServer.Run(); err != nil {
|
|
|
+ log.Warn("admin server exit with error: %v", err)
|
|
|
+ }
|
|
|
+ }()
|
|
|
}
|
|
|
<-svr.ctx.Done()
|
|
|
- // service context may not be canceled by svr.Close(), we should call it here to release resources
|
|
|
- if atomic.LoadUint32(&svr.exit) == 0 {
|
|
|
- svr.Close()
|
|
|
- }
|
|
|
+ svr.stop()
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
func (svr *Service) keepControllerWorking() {
|
|
|
- xl := xlog.FromContextSafe(svr.ctx)
|
|
|
- maxDelayTime := 20 * time.Second
|
|
|
- delayTime := time.Second
|
|
|
-
|
|
|
- // if frpc reconnect frps, we need to limit retry times in 1min
|
|
|
- // current retry logic is sleep 0s, 0s, 0s, 1s, 2s, 4s, 8s, ...
|
|
|
- // when exceed 1min, we will reset delay and counts
|
|
|
- cutoffTime := time.Now().Add(time.Minute)
|
|
|
- reconnectDelay := time.Second
|
|
|
- reconnectCounts := 1
|
|
|
-
|
|
|
- for {
|
|
|
- <-svr.ctl.ClosedDoneCh()
|
|
|
- if atomic.LoadUint32(&svr.exit) != 0 {
|
|
|
- return
|
|
|
- }
|
|
|
-
|
|
|
- // the first three attempts with a low delay
|
|
|
- if reconnectCounts > 3 {
|
|
|
- util.RandomSleep(reconnectDelay, 0.9, 1.1)
|
|
|
- xl.Info("wait %v to reconnect", reconnectDelay)
|
|
|
- reconnectDelay *= 2
|
|
|
- } else {
|
|
|
- util.RandomSleep(time.Second, 0, 0.5)
|
|
|
- }
|
|
|
- reconnectCounts++
|
|
|
-
|
|
|
- now := time.Now()
|
|
|
- if now.After(cutoffTime) {
|
|
|
- // reset
|
|
|
- cutoffTime = now.Add(time.Minute)
|
|
|
- reconnectDelay = time.Second
|
|
|
- reconnectCounts = 1
|
|
|
- }
|
|
|
-
|
|
|
- for {
|
|
|
- if atomic.LoadUint32(&svr.exit) != 0 {
|
|
|
- return
|
|
|
- }
|
|
|
-
|
|
|
- xl.Info("try to reconnect to server...")
|
|
|
- conn, cm, err := svr.login()
|
|
|
- if err != nil {
|
|
|
- xl.Warn("reconnect to server error: %v, wait %v for another retry", err, delayTime)
|
|
|
- util.RandomSleep(delayTime, 0.9, 1.1)
|
|
|
-
|
|
|
- delayTime *= 2
|
|
|
- if delayTime > maxDelayTime {
|
|
|
- delayTime = maxDelayTime
|
|
|
- }
|
|
|
- continue
|
|
|
- }
|
|
|
- // reconnect success, init delayTime
|
|
|
- delayTime = time.Second
|
|
|
-
|
|
|
- ctl := NewControl(svr.ctx, svr.runID, conn, cm, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.authSetter)
|
|
|
- ctl.Run()
|
|
|
- svr.ctlMu.Lock()
|
|
|
- if svr.ctl != nil {
|
|
|
- svr.ctl.Close()
|
|
|
- }
|
|
|
- svr.ctl = ctl
|
|
|
- svr.ctlMu.Unlock()
|
|
|
- break
|
|
|
+ <-svr.ctl.Done()
|
|
|
+
|
|
|
+ // There is a situation where the login is successful but due to certain reasons,
|
|
|
+ // the control immediately exits. It is necessary to limit the frequency of reconnection in this case.
|
|
|
+ // The interval for the first three retries in 1 minute will be very short, and then it will increase exponentially.
|
|
|
+ // The maximum interval is 20 seconds.
|
|
|
+ wait.BackoffUntil(func() error {
|
|
|
+ // loopLoginUntilSuccess is another layer of loop that will continuously attempt to
|
|
|
+ // login to the server until successful.
|
|
|
+ svr.loopLoginUntilSuccess(20*time.Second, false)
|
|
|
+ if svr.ctl != nil {
|
|
|
+ <-svr.ctl.Done()
|
|
|
+ return errors.New("control is closed and try another loop")
|
|
|
}
|
|
|
- }
|
|
|
+ // If the control is nil, it means that the login failed and the service is also closed.
|
|
|
+ return nil
|
|
|
+ }, wait.NewFastBackoffManager(
|
|
|
+ wait.FastBackoffOptions{
|
|
|
+ Duration: time.Second,
|
|
|
+ Factor: 2,
|
|
|
+ Jitter: 0.1,
|
|
|
+ MaxDuration: 20 * time.Second,
|
|
|
+ FastRetryCount: 3,
|
|
|
+ FastRetryDelay: 200 * time.Millisecond,
|
|
|
+ FastRetryWindow: time.Minute,
|
|
|
+ FastRetryJitter: 0.5,
|
|
|
+ },
|
|
|
+ ), true, svr.ctx.Done())
|
|
|
}
|
|
|
|
|
|
// login creates a connection to frps and registers it self as a client
|
|
|
// conn: control connection
|
|
|
// session: if it's not nil, using tcp mux
|
|
|
-func (svr *Service) login() (conn net.Conn, cm *ConnectionManager, err error) {
|
|
|
+func (svr *Service) login() (conn net.Conn, connector Connector, err error) {
|
|
|
xl := xlog.FromContextSafe(svr.ctx)
|
|
|
- cm = NewConnectionManager(svr.ctx, svr.cfg)
|
|
|
-
|
|
|
- if err = cm.OpenConnection(); err != nil {
|
|
|
+ connector = svr.connectorCreator(svr.ctx, svr.common)
|
|
|
+ if err = connector.Open(); err != nil {
|
|
|
return nil, nil, err
|
|
|
}
|
|
|
|
|
|
defer func() {
|
|
|
if err != nil {
|
|
|
- cm.Close()
|
|
|
+ connector.Close()
|
|
|
}
|
|
|
}()
|
|
|
|
|
|
- conn, err = cm.Connect()
|
|
|
+ conn, err = connector.Connect()
|
|
|
if err != nil {
|
|
|
return
|
|
|
}
|
|
@@ -262,12 +240,15 @@ func (svr *Service) login() (conn net.Conn, cm *ConnectionManager, err error) {
|
|
|
loginMsg := &msg.Login{
|
|
|
Arch: runtime.GOARCH,
|
|
|
Os: runtime.GOOS,
|
|
|
- PoolCount: svr.cfg.Transport.PoolCount,
|
|
|
- User: svr.cfg.User,
|
|
|
+ PoolCount: svr.common.Transport.PoolCount,
|
|
|
+ User: svr.common.User,
|
|
|
Version: version.Full(),
|
|
|
Timestamp: time.Now().Unix(),
|
|
|
RunID: svr.runID,
|
|
|
- Metas: svr.cfg.Metadatas,
|
|
|
+ Metas: svr.common.Metadatas,
|
|
|
+ }
|
|
|
+ if svr.clientSpec != nil {
|
|
|
+ loginMsg.ClientSpec = *svr.clientSpec
|
|
|
}
|
|
|
|
|
|
// Add auth
|
|
@@ -293,16 +274,79 @@ func (svr *Service) login() (conn net.Conn, cm *ConnectionManager, err error) {
|
|
|
}
|
|
|
|
|
|
svr.runID = loginRespMsg.RunID
|
|
|
- xl.ResetPrefixes()
|
|
|
- xl.AppendPrefix(svr.runID)
|
|
|
+ xl.AddPrefix(xlog.LogPrefix{Name: "runID", Value: svr.runID})
|
|
|
|
|
|
xl.Info("login to server success, get run id [%s]", loginRespMsg.RunID)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
-func (svr *Service) ReloadConf(pxyCfgs []v1.ProxyConfigurer, visitorCfgs []v1.VisitorConfigurer) error {
|
|
|
+func (svr *Service) loopLoginUntilSuccess(maxInterval time.Duration, firstLoginExit bool) {
|
|
|
+ xl := xlog.FromContextSafe(svr.ctx)
|
|
|
+ successCh := make(chan struct{})
|
|
|
+
|
|
|
+ loginFunc := func() error {
|
|
|
+ xl.Info("try to connect to server...")
|
|
|
+ conn, connector, err := svr.login()
|
|
|
+ if err != nil {
|
|
|
+ xl.Warn("connect to server error: %v", err)
|
|
|
+ if firstLoginExit {
|
|
|
+ svr.cancel(cancelErr{Err: err})
|
|
|
+ }
|
|
|
+ return err
|
|
|
+ }
|
|
|
+
|
|
|
+ svr.cfgMu.RLock()
|
|
|
+ proxyCfgs := svr.proxyCfgs
|
|
|
+ visitorCfgs := svr.visitorCfgs
|
|
|
+ svr.cfgMu.RUnlock()
|
|
|
+ connEncrypted := true
|
|
|
+ if svr.clientSpec != nil && svr.clientSpec.Type == "ssh-tunnel" {
|
|
|
+ connEncrypted = false
|
|
|
+ }
|
|
|
+ sessionCtx := &SessionContext{
|
|
|
+ Common: svr.common,
|
|
|
+ RunID: svr.runID,
|
|
|
+ Conn: conn,
|
|
|
+ ConnEncrypted: connEncrypted,
|
|
|
+ AuthSetter: svr.authSetter,
|
|
|
+ Connector: connector,
|
|
|
+ }
|
|
|
+ ctl, err := NewControl(svr.ctx, sessionCtx)
|
|
|
+ if err != nil {
|
|
|
+ conn.Close()
|
|
|
+ xl.Error("NewControl error: %v", err)
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ ctl.SetInWorkConnCallback(svr.handleWorkConnCb)
|
|
|
+
|
|
|
+ ctl.Run(proxyCfgs, visitorCfgs)
|
|
|
+ // close and replace previous control
|
|
|
+ svr.ctlMu.Lock()
|
|
|
+ if svr.ctl != nil {
|
|
|
+ svr.ctl.Close()
|
|
|
+ }
|
|
|
+ svr.ctl = ctl
|
|
|
+ svr.ctlMu.Unlock()
|
|
|
+
|
|
|
+ close(successCh)
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+
|
|
|
+ // try to reconnect to server until success
|
|
|
+ wait.BackoffUntil(loginFunc, wait.NewFastBackoffManager(
|
|
|
+ wait.FastBackoffOptions{
|
|
|
+ Duration: time.Second,
|
|
|
+ Factor: 2,
|
|
|
+ Jitter: 0.1,
|
|
|
+ MaxDuration: maxInterval,
|
|
|
+ }),
|
|
|
+ true,
|
|
|
+ wait.MergeAndCloseOnAnyStopChannel(svr.ctx.Done(), successCh))
|
|
|
+}
|
|
|
+
|
|
|
+func (svr *Service) UpdateAllConfigurer(proxyCfgs []v1.ProxyConfigurer, visitorCfgs []v1.VisitorConfigurer) error {
|
|
|
svr.cfgMu.Lock()
|
|
|
- svr.pxyCfgs = pxyCfgs
|
|
|
+ svr.proxyCfgs = proxyCfgs
|
|
|
svr.visitorCfgs = visitorCfgs
|
|
|
svr.cfgMu.Unlock()
|
|
|
|
|
@@ -311,7 +355,7 @@ func (svr *Service) ReloadConf(pxyCfgs []v1.ProxyConfigurer, visitorCfgs []v1.Vi
|
|
|
svr.ctlMu.RUnlock()
|
|
|
|
|
|
if ctl != nil {
|
|
|
- return svr.ctl.ReloadConf(pxyCfgs, visitorCfgs)
|
|
|
+ return svr.ctl.UpdateAllConfigurer(proxyCfgs, visitorCfgs)
|
|
|
}
|
|
|
return nil
|
|
|
}
|
|
@@ -321,191 +365,31 @@ func (svr *Service) Close() {
|
|
|
}
|
|
|
|
|
|
func (svr *Service) GracefulClose(d time.Duration) {
|
|
|
- atomic.StoreUint32(&svr.exit, 1)
|
|
|
+ svr.gracefulShutdownDuration = d
|
|
|
+ svr.cancel(nil)
|
|
|
+}
|
|
|
|
|
|
- svr.ctlMu.RLock()
|
|
|
+func (svr *Service) stop() {
|
|
|
+ svr.ctlMu.Lock()
|
|
|
+ defer svr.ctlMu.Unlock()
|
|
|
if svr.ctl != nil {
|
|
|
- svr.ctl.GracefulClose(d)
|
|
|
+ svr.ctl.GracefulClose(svr.gracefulShutdownDuration)
|
|
|
svr.ctl = nil
|
|
|
}
|
|
|
- svr.ctlMu.RUnlock()
|
|
|
-
|
|
|
- if svr.cancel != nil {
|
|
|
- svr.cancel()
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-type ConnectionManager struct {
|
|
|
- ctx context.Context
|
|
|
- cfg *v1.ClientCommonConfig
|
|
|
-
|
|
|
- muxSession *fmux.Session
|
|
|
- quicConn quic.Connection
|
|
|
-}
|
|
|
-
|
|
|
-func NewConnectionManager(ctx context.Context, cfg *v1.ClientCommonConfig) *ConnectionManager {
|
|
|
- return &ConnectionManager{
|
|
|
- ctx: ctx,
|
|
|
- cfg: cfg,
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-func (cm *ConnectionManager) OpenConnection() error {
|
|
|
- xl := xlog.FromContextSafe(cm.ctx)
|
|
|
-
|
|
|
- // special for quic
|
|
|
- if strings.EqualFold(cm.cfg.Transport.Protocol, "quic") {
|
|
|
- var tlsConfig *tls.Config
|
|
|
- var err error
|
|
|
- sn := cm.cfg.Transport.TLS.ServerName
|
|
|
- if sn == "" {
|
|
|
- sn = cm.cfg.ServerAddr
|
|
|
- }
|
|
|
- if lo.FromPtr(cm.cfg.Transport.TLS.Enable) {
|
|
|
- tlsConfig, err = transport.NewClientTLSConfig(
|
|
|
- cm.cfg.Transport.TLS.CertFile,
|
|
|
- cm.cfg.Transport.TLS.KeyFile,
|
|
|
- cm.cfg.Transport.TLS.TrustedCaFile,
|
|
|
- sn)
|
|
|
- } else {
|
|
|
- tlsConfig, err = transport.NewClientTLSConfig("", "", "", sn)
|
|
|
- }
|
|
|
- if err != nil {
|
|
|
- xl.Warn("fail to build tls configuration, err: %v", err)
|
|
|
- return err
|
|
|
- }
|
|
|
- tlsConfig.NextProtos = []string{"frp"}
|
|
|
-
|
|
|
- conn, err := quic.DialAddr(
|
|
|
- cm.ctx,
|
|
|
- net.JoinHostPort(cm.cfg.ServerAddr, strconv.Itoa(cm.cfg.ServerPort)),
|
|
|
- tlsConfig, &quic.Config{
|
|
|
- MaxIdleTimeout: time.Duration(cm.cfg.Transport.QUIC.MaxIdleTimeout) * time.Second,
|
|
|
- MaxIncomingStreams: int64(cm.cfg.Transport.QUIC.MaxIncomingStreams),
|
|
|
- KeepAlivePeriod: time.Duration(cm.cfg.Transport.QUIC.KeepalivePeriod) * time.Second,
|
|
|
- })
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- cm.quicConn = conn
|
|
|
- return nil
|
|
|
- }
|
|
|
-
|
|
|
- if !lo.FromPtr(cm.cfg.Transport.TCPMux) {
|
|
|
- return nil
|
|
|
- }
|
|
|
-
|
|
|
- conn, err := cm.realConnect()
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
-
|
|
|
- fmuxCfg := fmux.DefaultConfig()
|
|
|
- fmuxCfg.KeepAliveInterval = time.Duration(cm.cfg.Transport.TCPMuxKeepaliveInterval) * time.Second
|
|
|
- fmuxCfg.LogOutput = io.Discard
|
|
|
- fmuxCfg.MaxStreamWindowSize = 6 * 1024 * 1024
|
|
|
- session, err := fmux.Client(conn, fmuxCfg)
|
|
|
- if err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
- cm.muxSession = session
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (cm *ConnectionManager) Connect() (net.Conn, error) {
|
|
|
- if cm.quicConn != nil {
|
|
|
- stream, err := cm.quicConn.OpenStreamSync(context.Background())
|
|
|
- if err != nil {
|
|
|
- return nil, err
|
|
|
- }
|
|
|
- return utilnet.QuicStreamToNetConn(stream, cm.quicConn), nil
|
|
|
- } else if cm.muxSession != nil {
|
|
|
- stream, err := cm.muxSession.OpenStream()
|
|
|
- if err != nil {
|
|
|
- return nil, err
|
|
|
- }
|
|
|
- return stream, nil
|
|
|
- }
|
|
|
-
|
|
|
- return cm.realConnect()
|
|
|
}
|
|
|
|
|
|
-func (cm *ConnectionManager) realConnect() (net.Conn, error) {
|
|
|
- xl := xlog.FromContextSafe(cm.ctx)
|
|
|
- var tlsConfig *tls.Config
|
|
|
- var err error
|
|
|
- tlsEnable := lo.FromPtr(cm.cfg.Transport.TLS.Enable)
|
|
|
- if cm.cfg.Transport.Protocol == "wss" {
|
|
|
- tlsEnable = true
|
|
|
- }
|
|
|
- if tlsEnable {
|
|
|
- sn := cm.cfg.Transport.TLS.ServerName
|
|
|
- if sn == "" {
|
|
|
- sn = cm.cfg.ServerAddr
|
|
|
- }
|
|
|
-
|
|
|
- tlsConfig, err = transport.NewClientTLSConfig(
|
|
|
- cm.cfg.Transport.TLS.CertFile,
|
|
|
- cm.cfg.Transport.TLS.KeyFile,
|
|
|
- cm.cfg.Transport.TLS.TrustedCaFile,
|
|
|
- sn)
|
|
|
- if err != nil {
|
|
|
- xl.Warn("fail to build tls configuration, err: %v", err)
|
|
|
- return nil, err
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- proxyType, addr, auth, err := libdial.ParseProxyURL(cm.cfg.Transport.ProxyURL)
|
|
|
- if err != nil {
|
|
|
- xl.Error("fail to parse proxy url")
|
|
|
- return nil, err
|
|
|
- }
|
|
|
- dialOptions := []libdial.DialOption{}
|
|
|
- protocol := cm.cfg.Transport.Protocol
|
|
|
- switch protocol {
|
|
|
- case "websocket":
|
|
|
- protocol = "tcp"
|
|
|
- dialOptions = append(dialOptions, libdial.WithAfterHook(libdial.AfterHook{Hook: utilnet.DialHookWebsocket(protocol, "")}))
|
|
|
- dialOptions = append(dialOptions, libdial.WithAfterHook(libdial.AfterHook{
|
|
|
- Hook: utilnet.DialHookCustomTLSHeadByte(tlsConfig != nil, lo.FromPtr(cm.cfg.Transport.TLS.DisableCustomTLSFirstByte)),
|
|
|
- }))
|
|
|
- dialOptions = append(dialOptions, libdial.WithTLSConfig(tlsConfig))
|
|
|
- case "wss":
|
|
|
- protocol = "tcp"
|
|
|
- dialOptions = append(dialOptions, libdial.WithTLSConfigAndPriority(100, tlsConfig))
|
|
|
- // Make sure that if it is wss, the websocket hook is executed after the tls hook.
|
|
|
- dialOptions = append(dialOptions, libdial.WithAfterHook(libdial.AfterHook{Hook: utilnet.DialHookWebsocket(protocol, tlsConfig.ServerName), Priority: 110}))
|
|
|
- default:
|
|
|
- dialOptions = append(dialOptions, libdial.WithAfterHook(libdial.AfterHook{
|
|
|
- Hook: utilnet.DialHookCustomTLSHeadByte(tlsConfig != nil, lo.FromPtr(cm.cfg.Transport.TLS.DisableCustomTLSFirstByte)),
|
|
|
- }))
|
|
|
- dialOptions = append(dialOptions, libdial.WithTLSConfig(tlsConfig))
|
|
|
- }
|
|
|
-
|
|
|
- if cm.cfg.Transport.ConnectServerLocalIP != "" {
|
|
|
- dialOptions = append(dialOptions, libdial.WithLocalAddr(cm.cfg.Transport.ConnectServerLocalIP))
|
|
|
- }
|
|
|
- dialOptions = append(dialOptions,
|
|
|
- libdial.WithProtocol(protocol),
|
|
|
- libdial.WithTimeout(time.Duration(cm.cfg.Transport.DialServerTimeout)*time.Second),
|
|
|
- libdial.WithKeepAlive(time.Duration(cm.cfg.Transport.DialServerKeepAlive)*time.Second),
|
|
|
- libdial.WithProxy(proxyType, addr),
|
|
|
- libdial.WithProxyAuth(auth),
|
|
|
- )
|
|
|
- conn, err := libdial.DialContext(
|
|
|
- cm.ctx,
|
|
|
- net.JoinHostPort(cm.cfg.ServerAddr, strconv.Itoa(cm.cfg.ServerPort)),
|
|
|
- dialOptions...,
|
|
|
- )
|
|
|
- return conn, err
|
|
|
-}
|
|
|
+// TODO(fatedier): Use StatusExporter to provide query interfaces instead of directly using methods from the Service.
|
|
|
+func (svr *Service) GetProxyStatus(name string) (*proxy.WorkingStatus, error) {
|
|
|
+ svr.ctlMu.RLock()
|
|
|
+ ctl := svr.ctl
|
|
|
+ svr.ctlMu.RUnlock()
|
|
|
|
|
|
-func (cm *ConnectionManager) Close() error {
|
|
|
- if cm.quicConn != nil {
|
|
|
- _ = cm.quicConn.CloseWithError(0, "")
|
|
|
+ if ctl == nil {
|
|
|
+ return nil, fmt.Errorf("control is not running")
|
|
|
}
|
|
|
- if cm.muxSession != nil {
|
|
|
- _ = cm.muxSession.Close()
|
|
|
+ ws, ok := ctl.pm.GetProxyStatus(name)
|
|
|
+ if !ok {
|
|
|
+ return nil, fmt.Errorf("proxy [%s] is not found", name)
|
|
|
}
|
|
|
- return nil
|
|
|
+ return ws, nil
|
|
|
}
|