frp/server/control.go

664 lines
16 KiB
Go
Raw Normal View History

2017-03-23 02:01:25 +08:00
// Copyright 2017 fatedier, fatedier@gmail.com
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
2017-03-09 02:03:47 +08:00
package server
import (
2019-10-12 20:13:12 +08:00
"context"
2017-03-09 02:03:47 +08:00
"fmt"
"io"
2019-10-12 20:13:12 +08:00
"net"
2018-05-11 10:42:57 +08:00
"runtime/debug"
2017-03-09 02:03:47 +08:00
"sync"
"time"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/golib/control/shutdown"
"github.com/fatedier/golib/crypto"
"github.com/fatedier/golib/errors"
"github.com/samber/lo"
2022-08-29 01:02:53 +08:00
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/auth"
"github.com/fatedier/frp/pkg/config"
v1 "github.com/fatedier/frp/pkg/config/v1"
2023-05-29 14:10:34 +08:00
pkgerr "github.com/fatedier/frp/pkg/errors"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/msg"
plugin "github.com/fatedier/frp/pkg/plugin/server"
"github.com/fatedier/frp/pkg/transport"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/version"
"github.com/fatedier/frp/pkg/util/xlog"
2019-01-15 00:11:08 +08:00
"github.com/fatedier/frp/server/controller"
"github.com/fatedier/frp/server/metrics"
2019-01-15 00:11:08 +08:00
"github.com/fatedier/frp/server/proxy"
2017-03-09 02:03:47 +08:00
)
2019-01-15 00:11:08 +08:00
type ControlManager struct {
// controls indexed by run id
2020-05-24 17:48:37 +08:00
ctlsByRunID map[string]*Control
2019-01-15 00:11:08 +08:00
mu sync.RWMutex
}
func NewControlManager() *ControlManager {
return &ControlManager{
2020-05-24 17:48:37 +08:00
ctlsByRunID: make(map[string]*Control),
2019-01-15 00:11:08 +08:00
}
}
func (cm *ControlManager) Add(runID string, ctl *Control) (old *Control) {
2019-01-15 00:11:08 +08:00
cm.mu.Lock()
defer cm.mu.Unlock()
var ok bool
old, ok = cm.ctlsByRunID[runID]
2019-01-15 00:11:08 +08:00
if ok {
old.Replaced(ctl)
2019-01-15 00:11:08 +08:00
}
2020-05-24 17:48:37 +08:00
cm.ctlsByRunID[runID] = ctl
2019-01-15 00:11:08 +08:00
return
}
2019-01-26 21:36:24 +08:00
// we should make sure if it's the same control to prevent delete a new one
2020-05-24 17:48:37 +08:00
func (cm *ControlManager) Del(runID string, ctl *Control) {
2019-01-15 00:11:08 +08:00
cm.mu.Lock()
defer cm.mu.Unlock()
2020-05-24 17:48:37 +08:00
if c, ok := cm.ctlsByRunID[runID]; ok && c == ctl {
delete(cm.ctlsByRunID, runID)
2019-01-26 21:36:24 +08:00
}
2019-01-15 00:11:08 +08:00
}
2020-05-24 17:48:37 +08:00
func (cm *ControlManager) GetByID(runID string) (ctl *Control, ok bool) {
2019-01-15 00:11:08 +08:00
cm.mu.RLock()
defer cm.mu.RUnlock()
2020-05-24 17:48:37 +08:00
ctl, ok = cm.ctlsByRunID[runID]
2019-01-15 00:11:08 +08:00
return
}
func (cm *ControlManager) Close() error {
cm.mu.Lock()
defer cm.mu.Unlock()
for _, ctl := range cm.ctlsByRunID {
ctl.Close()
}
cm.ctlsByRunID = make(map[string]*Control)
return nil
}
2017-03-09 02:03:47 +08:00
type Control struct {
2019-01-10 20:53:06 +08:00
// all resource managers and controllers
2019-01-15 00:11:08 +08:00
rc *controller.ResourceController
// proxy manager
2020-05-24 17:48:37 +08:00
pxyManager *proxy.Manager
2019-01-15 00:11:08 +08:00
2019-12-20 20:28:28 +08:00
// plugin manager
pluginManager *plugin.Manager
// verifies authentication based on selected method
authVerifier auth.Verifier
// other components can use this to communicate with client
msgTransporter transport.MessageTransporter
2017-03-09 02:03:47 +08:00
// login message
loginMsg *msg.Login
// control connection
conn net.Conn
// put a message in this channel to send it over control connection to client
sendCh chan (msg.Message)
// read from this channel to get the next message sent by client
readCh chan (msg.Message)
// work connections
workConnCh chan net.Conn
// proxies in one client
2019-01-15 00:11:08 +08:00
proxies map[string]proxy.Proxy
2017-03-09 02:03:47 +08:00
// pool count
poolCount int
2018-01-26 14:56:55 +08:00
// ports used, for limitations
portsUsedNum int
2017-03-09 02:03:47 +08:00
// last time got the Ping message
lastPing time.Time
// A new run id will be generated when a new client login.
// If run id got from login message has same run id, it means it's the same client, so we can
// replace old controller instantly.
2020-05-24 17:48:37 +08:00
runID string
2017-03-09 02:03:47 +08:00
readerShutdown *shutdown.Shutdown
writerShutdown *shutdown.Shutdown
managerShutdown *shutdown.Shutdown
allShutdown *shutdown.Shutdown
started bool
2017-03-09 02:03:47 +08:00
mu sync.RWMutex
// Server configuration information
serverCfg *v1.ServerConfig
2019-10-12 20:13:12 +08:00
xl *xlog.Logger
ctx context.Context
2017-03-09 02:03:47 +08:00
}
2019-12-20 20:28:28 +08:00
func NewControl(
ctx context.Context,
rc *controller.ResourceController,
2020-05-24 17:48:37 +08:00
pxyManager *proxy.Manager,
2019-12-20 20:28:28 +08:00
pluginManager *plugin.Manager,
authVerifier auth.Verifier,
2019-12-20 20:28:28 +08:00
ctlConn net.Conn,
loginMsg *msg.Login,
serverCfg *v1.ServerConfig,
2019-12-20 20:28:28 +08:00
) *Control {
2019-08-29 21:13:21 +08:00
poolCount := loginMsg.PoolCount
if poolCount > int(serverCfg.Transport.MaxPoolCount) {
poolCount = int(serverCfg.Transport.MaxPoolCount)
2019-08-29 21:13:21 +08:00
}
ctl := &Control{
2019-01-10 20:53:06 +08:00
rc: rc,
2019-01-15 00:11:08 +08:00
pxyManager: pxyManager,
2019-12-20 20:28:28 +08:00
pluginManager: pluginManager,
authVerifier: authVerifier,
2017-03-09 02:03:47 +08:00
conn: ctlConn,
loginMsg: loginMsg,
sendCh: make(chan msg.Message, 10),
readCh: make(chan msg.Message, 10),
2019-08-29 21:13:21 +08:00
workConnCh: make(chan net.Conn, poolCount+10),
2019-01-15 00:11:08 +08:00
proxies: make(map[string]proxy.Proxy),
2019-08-29 21:13:21 +08:00
poolCount: poolCount,
2018-01-26 14:56:55 +08:00
portsUsedNum: 0,
2017-03-09 02:03:47 +08:00
lastPing: time.Now(),
2020-05-24 17:48:37 +08:00
runID: loginMsg.RunID,
2017-03-09 02:03:47 +08:00
readerShutdown: shutdown.New(),
writerShutdown: shutdown.New(),
managerShutdown: shutdown.New(),
allShutdown: shutdown.New(),
serverCfg: serverCfg,
2019-10-12 20:13:12 +08:00
xl: xlog.FromContextSafe(ctx),
ctx: ctx,
2017-03-09 02:03:47 +08:00
}
ctl.msgTransporter = transport.NewMessageTransporter(ctl.sendCh)
return ctl
2017-03-09 02:03:47 +08:00
}
// Start send a login success message to client and start working.
func (ctl *Control) Start() {
2017-03-10 01:42:06 +08:00
loginRespMsg := &msg.LoginResp{
2023-05-29 01:09:53 +08:00
Version: version.Full(),
RunID: ctl.runID,
Error: "",
2017-03-09 02:03:47 +08:00
}
2022-08-29 01:02:53 +08:00
_ = msg.WriteMsg(ctl.conn, loginRespMsg)
ctl.mu.Lock()
ctl.started = true
ctl.mu.Unlock()
2017-03-09 02:03:47 +08:00
2017-03-10 01:42:06 +08:00
go ctl.writer()
go func() {
for i := 0; i < ctl.poolCount; i++ {
// ignore error here, that means that this control is closed
_ = errors.PanicToError(func() {
ctl.sendCh <- &msg.ReqWorkConn{}
})
}
}()
2017-03-09 02:03:47 +08:00
go ctl.manager()
go ctl.reader()
go ctl.stoper()
}
func (ctl *Control) Close() error {
ctl.allShutdown.Start()
return nil
}
func (ctl *Control) Replaced(newCtl *Control) {
xl := ctl.xl
xl.Info("Replaced by client [%s]", newCtl.runID)
ctl.runID = ""
ctl.allShutdown.Start()
}
func (ctl *Control) RegisterWorkConn(conn net.Conn) error {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
select {
case ctl.workConnCh <- conn:
2019-10-12 20:13:12 +08:00
xl.Debug("new work connection registered")
return nil
2017-03-09 02:03:47 +08:00
default:
2019-10-12 20:13:12 +08:00
xl.Debug("work connection pool is full, discarding")
return fmt.Errorf("work connection pool is full, discarding")
2017-03-09 02:03:47 +08:00
}
}
// When frps get one user connection, we get one work connection from the pool and return it.
// If no workConn available in the pool, send message to frpc to get one or more
// and wait until it is available.
// return an error if wait timeout
func (ctl *Control) GetWorkConn() (workConn net.Conn, err error) {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
var ok bool
// get a work connection from the pool
select {
case workConn, ok = <-ctl.workConnCh:
if !ok {
2023-05-29 14:10:34 +08:00
err = pkgerr.ErrCtlClosed
2017-03-09 02:03:47 +08:00
return
}
2019-10-12 20:13:12 +08:00
xl.Debug("get work connection from pool")
2017-03-09 02:03:47 +08:00
default:
// no work connections available in the poll, send message to frpc to get more
if err = errors.PanicToError(func() {
2017-03-09 02:03:47 +08:00
ctl.sendCh <- &msg.ReqWorkConn{}
}); err != nil {
return nil, fmt.Errorf("control is already closed")
2017-03-09 02:03:47 +08:00
}
select {
case workConn, ok = <-ctl.workConnCh:
if !ok {
2023-05-29 14:10:34 +08:00
err = pkgerr.ErrCtlClosed
xl.Warn("no work connections available, %v", err)
2017-03-09 02:03:47 +08:00
return
}
case <-time.After(time.Duration(ctl.serverCfg.UserConnTimeout) * time.Second):
2017-03-09 02:03:47 +08:00
err = fmt.Errorf("timeout trying to get work connection")
2019-10-12 20:13:12 +08:00
xl.Warn("%v", err)
2017-03-09 02:03:47 +08:00
return
}
}
// When we get a work connection from pool, replace it with a new one.
2022-08-29 01:02:53 +08:00
_ = errors.PanicToError(func() {
2017-03-09 02:03:47 +08:00
ctl.sendCh <- &msg.ReqWorkConn{}
})
return
}
func (ctl *Control) writer() {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
defer ctl.allShutdown.Start()
defer ctl.writerShutdown.Done()
encWriter, err := crypto.NewWriter(ctl.conn, []byte(ctl.serverCfg.Auth.Token))
2017-03-10 01:42:06 +08:00
if err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("crypto new writer error: %v", err)
2017-03-10 01:42:06 +08:00
ctl.allShutdown.Start()
return
}
2017-03-09 02:03:47 +08:00
for {
2020-05-24 17:48:37 +08:00
m, ok := <-ctl.sendCh
if !ok {
2019-10-12 20:13:12 +08:00
xl.Info("control writer is closing")
2017-03-09 02:03:47 +08:00
return
2020-05-24 17:48:37 +08:00
}
if err := msg.WriteMsg(encWriter, m); err != nil {
xl.Warn("write message to control connection error: %v", err)
return
2017-03-09 02:03:47 +08:00
}
}
}
func (ctl *Control) reader() {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
defer ctl.allShutdown.Start()
defer ctl.readerShutdown.Done()
encReader := crypto.NewReader(ctl.conn, []byte(ctl.serverCfg.Auth.Token))
2017-03-09 02:03:47 +08:00
for {
2020-05-24 17:48:37 +08:00
m, err := msg.ReadMsg(encReader)
if err != nil {
2017-03-09 02:03:47 +08:00
if err == io.EOF {
2019-10-12 20:13:12 +08:00
xl.Debug("control connection closed")
2017-03-09 02:03:47 +08:00
return
}
2020-05-24 17:48:37 +08:00
xl.Warn("read error: %v", err)
ctl.conn.Close()
return
2017-03-09 02:03:47 +08:00
}
2020-05-24 17:48:37 +08:00
ctl.readCh <- m
2017-03-09 02:03:47 +08:00
}
}
func (ctl *Control) stoper() {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
ctl.allShutdown.WaitStart()
ctl.conn.Close()
ctl.readerShutdown.WaitDone()
2017-03-09 02:03:47 +08:00
close(ctl.readCh)
2018-01-17 01:09:33 +08:00
ctl.managerShutdown.WaitDone()
2017-03-09 02:03:47 +08:00
close(ctl.sendCh)
2018-01-17 01:09:33 +08:00
ctl.writerShutdown.WaitDone()
2017-03-09 02:03:47 +08:00
2018-03-19 20:22:15 +08:00
ctl.mu.Lock()
defer ctl.mu.Unlock()
2017-03-09 02:03:47 +08:00
close(ctl.workConnCh)
for workConn := range ctl.workConnCh {
workConn.Close()
}
for _, pxy := range ctl.proxies {
pxy.Close()
2019-01-15 00:11:08 +08:00
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() {
2022-08-29 01:02:53 +08:00
_ = ctl.pluginManager.CloseProxy(notifyContent)
}()
2017-03-09 02:03:47 +08:00
}
ctl.allShutdown.Done()
2019-10-12 20:13:12 +08:00
xl.Info("client exit success")
metrics.Server.CloseClient()
2019-01-15 00:11:08 +08:00
}
// block until Control closed
func (ctl *Control) WaitClosed() {
ctl.mu.RLock()
started := ctl.started
ctl.mu.RUnlock()
if !started {
ctl.allShutdown.Done()
return
}
2019-01-15 00:11:08 +08:00
ctl.allShutdown.WaitDone()
2017-03-09 02:03:47 +08:00
}
func (ctl *Control) manager() {
2019-10-12 20:13:12 +08:00
xl := ctl.xl
2017-03-09 02:03:47 +08:00
defer func() {
if err := recover(); err != nil {
2019-10-12 20:13:12 +08:00
xl.Error("panic error: %v", err)
xl.Error(string(debug.Stack()))
2017-03-09 02:03:47 +08:00
}
}()
defer ctl.allShutdown.Start()
defer ctl.managerShutdown.Done()
var heartbeatCh <-chan time.Time
2023-07-21 10:30:46 +08:00
// Don't need application heartbeat if TCPMux is enabled,
// yamux will do same thing.
if !lo.FromPtr(ctl.serverCfg.Transport.TCPMux) && ctl.serverCfg.Transport.HeartbeatTimeout > 0 {
heartbeat := time.NewTicker(time.Second)
defer heartbeat.Stop()
heartbeatCh = heartbeat.C
}
2017-03-09 02:03:47 +08:00
for {
select {
case <-heartbeatCh:
if time.Since(ctl.lastPing) > time.Duration(ctl.serverCfg.Transport.HeartbeatTimeout)*time.Second {
2019-10-12 20:13:12 +08:00
xl.Warn("heartbeat timeout")
2018-03-19 20:22:15 +08:00
return
2017-03-09 02:03:47 +08:00
}
case rawMsg, ok := <-ctl.readCh:
if !ok {
return
}
switch m := rawMsg.(type) {
case *msg.NewProxy:
2019-12-20 20:28:28 +08:00
content := &plugin.NewProxyContent{
User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
2020-05-24 17:48:37 +08:00
RunID: ctl.loginMsg.RunID,
2019-12-20 20:28:28 +08:00
},
NewProxy: *m,
}
var remoteAddr string
retContent, err := ctl.pluginManager.NewProxy(content)
if err == nil {
m = &retContent.NewProxy
remoteAddr, err = ctl.RegisterProxy(m)
}
2017-03-09 02:03:47 +08:00
// register proxy in this control
resp := &msg.NewProxyResp{
ProxyName: m.ProxyName,
}
if err != nil {
xl.Warn("new proxy [%s] type [%s] error: %v", m.ProxyName, m.ProxyType, err)
resp.Error = util.GenerateResponseErrorString(fmt.Sprintf("new proxy [%s] error", m.ProxyName),
err, lo.FromPtr(ctl.serverCfg.DetailedErrorsToClient))
2017-03-09 02:03:47 +08:00
} else {
2018-01-17 14:40:08 +08:00
resp.RemoteAddr = remoteAddr
xl.Info("new proxy [%s] type [%s] success", m.ProxyName, m.ProxyType)
metrics.Server.NewProxy(m.ProxyName, m.ProxyType)
2017-03-09 02:03:47 +08:00
}
ctl.sendCh <- resp
case *msg.NatHoleVisitor:
go ctl.HandleNatHoleVisitor(m)
case *msg.NatHoleClient:
go ctl.HandleNatHoleClient(m)
case *msg.NatHoleReport:
go ctl.HandleNatHoleReport(m)
2017-06-11 17:22:05 +08:00
case *msg.CloseProxy:
2022-08-29 01:02:53 +08:00
_ = ctl.CloseProxy(m)
2019-10-12 20:13:12 +08:00
xl.Info("close proxy [%s] success", m.ProxyName)
2017-03-09 02:03:47 +08:00
case *msg.Ping:
content := &plugin.PingContent{
User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
2020-05-24 17:48:37 +08:00
RunID: ctl.loginMsg.RunID,
},
Ping: *m,
}
retContent, err := ctl.pluginManager.Ping(content)
if err == nil {
m = &retContent.Ping
err = ctl.authVerifier.VerifyPing(m)
}
if err != nil {
xl.Warn("received invalid ping: %v", err)
ctl.sendCh <- &msg.Pong{
Error: util.GenerateResponseErrorString("invalid ping", err, lo.FromPtr(ctl.serverCfg.DetailedErrorsToClient)),
}
return
}
2017-03-09 02:03:47 +08:00
ctl.lastPing = time.Now()
2019-10-12 20:13:12 +08:00
xl.Debug("receive heartbeat")
2017-03-09 02:03:47 +08:00
ctl.sendCh <- &msg.Pong{}
}
}
}
}
func (ctl *Control) HandleNatHoleVisitor(m *msg.NatHoleVisitor) {
ctl.rc.NatHoleController.HandleVisitor(m, ctl.msgTransporter, ctl.loginMsg.User)
}
func (ctl *Control) HandleNatHoleClient(m *msg.NatHoleClient) {
ctl.rc.NatHoleController.HandleClient(m, ctl.msgTransporter)
}
func (ctl *Control) HandleNatHoleReport(m *msg.NatHoleReport) {
ctl.rc.NatHoleController.HandleReport(m)
}
2018-01-17 14:40:08 +08:00
func (ctl *Control) RegisterProxy(pxyMsg *msg.NewProxy) (remoteAddr string, err error) {
var pxyConf v1.ProxyConfigurer
// Load configures from NewProxy message and validate.
pxyConf, err = config.NewProxyConfigurerFromMsg(pxyMsg, ctl.serverCfg)
2017-03-09 02:03:47 +08:00
if err != nil {
2018-01-17 14:40:08 +08:00
return
2017-03-09 02:03:47 +08:00
}
// User info
userInfo := plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
2020-05-24 17:48:37 +08:00
RunID: ctl.runID,
}
// NewProxy will return an interface Proxy.
// In fact, it creates different proxies based on the proxy type. We just call run() here.
2023-09-20 15:18:50 +08:00
pxy, err := proxy.NewProxy(ctl.ctx, &proxy.Options{
UserInfo: userInfo,
LoginMsg: ctl.loginMsg,
PoolCount: ctl.poolCount,
ResourceController: ctl.rc,
GetWorkConnFn: ctl.GetWorkConn,
Configurer: pxyConf,
ServerCfg: ctl.serverCfg,
})
2017-03-09 02:03:47 +08:00
if err != nil {
2018-01-17 14:40:08 +08:00
return remoteAddr, err
2017-03-09 02:03:47 +08:00
}
2018-01-26 14:56:55 +08:00
// Check ports used number in each client
if ctl.serverCfg.MaxPortsPerClient > 0 {
2018-01-26 14:56:55 +08:00
ctl.mu.Lock()
if ctl.portsUsedNum+pxy.GetUsedPortsNum() > int(ctl.serverCfg.MaxPortsPerClient) {
2018-01-26 14:56:55 +08:00
ctl.mu.Unlock()
err = fmt.Errorf("exceed the max_ports_per_client")
return
}
2022-08-29 01:02:53 +08:00
ctl.portsUsedNum += pxy.GetUsedPortsNum()
2018-01-26 14:56:55 +08:00
ctl.mu.Unlock()
defer func() {
if err != nil {
ctl.mu.Lock()
2022-08-29 01:02:53 +08:00
ctl.portsUsedNum -= pxy.GetUsedPortsNum()
2018-01-26 14:56:55 +08:00
ctl.mu.Unlock()
}
}()
}
if ctl.pxyManager.Exist(pxyMsg.ProxyName) {
err = fmt.Errorf("proxy [%s] already exists", pxyMsg.ProxyName)
return
}
2018-01-17 14:40:08 +08:00
remoteAddr, err = pxy.Run()
2017-03-09 02:03:47 +08:00
if err != nil {
2018-01-17 14:40:08 +08:00
return
2017-03-09 02:03:47 +08:00
}
defer func() {
if err != nil {
pxy.Close()
}
}()
2019-01-15 00:11:08 +08:00
err = ctl.pxyManager.Add(pxyMsg.ProxyName, pxy)
2017-03-09 02:03:47 +08:00
if err != nil {
2018-01-17 14:40:08 +08:00
return
2017-03-09 02:03:47 +08:00
}
2017-06-11 17:22:05 +08:00
ctl.mu.Lock()
ctl.proxies[pxy.GetName()] = pxy
ctl.mu.Unlock()
2018-01-17 14:40:08 +08:00
return
2017-03-09 02:03:47 +08:00
}
2017-06-11 17:22:05 +08:00
func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
ctl.mu.Lock()
pxy, ok := ctl.proxies[closeMsg.ProxyName]
if !ok {
2018-01-26 14:56:55 +08:00
ctl.mu.Unlock()
2017-06-11 17:22:05 +08:00
return
}
if ctl.serverCfg.MaxPortsPerClient > 0 {
2022-08-29 01:02:53 +08:00
ctl.portsUsedNum -= pxy.GetUsedPortsNum()
2018-01-26 14:56:55 +08:00
}
2017-06-11 17:22:05 +08:00
pxy.Close()
2019-01-15 00:11:08 +08:00
ctl.pxyManager.Del(pxy.GetName())
2017-06-27 01:59:30 +08:00
delete(ctl.proxies, closeMsg.ProxyName)
2018-01-26 14:56:55 +08:00
ctl.mu.Unlock()
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() {
2022-08-29 01:02:53 +08:00
_ = ctl.pluginManager.CloseProxy(notifyContent)
}()
2017-06-11 17:22:05 +08:00
return
}