Merge pull request #2834 from fatedier/dev

bump version
pull/2856/head v0.40.0
fatedier 2022-03-11 19:51:32 +08:00 committed by GitHub
commit ce677820c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 232 additions and 32 deletions

View File

@ -1,3 +1,13 @@
### New
* Added `dial_server_timeout` in frpc to specify connect timeout to frps.
* Additional EndpointParams can be set for OIDC.
* Added CloseProxy operation in server plugin.
### Improve
* Added some randomness in reconnect delay.
### Fix ### Fix
* Fixed IPv6 address parse issue. * TLS server name is ignored when `tls_trusted_ca_file` isnt set.

View File

@ -251,6 +251,7 @@ func (ctl *Control) connectServer() (conn net.Conn, err error) {
} }
dialOptions = append(dialOptions, dialOptions = append(dialOptions,
libdial.WithProtocol(protocol), libdial.WithProtocol(protocol),
libdial.WithTimeout(time.Duration(ctl.clientCfg.DialServerTimeout)*time.Second),
libdial.WithProxy(proxyType, addr), libdial.WithProxy(proxyType, addr),
libdial.WithProxyAuth(auth), libdial.WithProxyAuth(auth),
libdial.WithTLSConfig(tlsConfig), libdial.WithTLSConfig(tlsConfig),

View File

@ -787,7 +787,10 @@ func HandleTCPWorkConnection(ctx context.Context, localInfo *config.LocalSvrConf
return return
} }
localConn, err := libdial.Dial(net.JoinHostPort(localInfo.LocalIP, strconv.Itoa(localInfo.LocalPort))) localConn, err := libdial.Dial(
net.JoinHostPort(localInfo.LocalIP, strconv.Itoa(localInfo.LocalPort)),
libdial.WithTimeout(10*time.Second),
)
if err != nil { if err != nil {
workConn.Close() workConn.Close()
xl.Error("connect to local service [%s:%d] error: %v", localInfo.LocalIP, localInfo.LocalPort, err) xl.Error("connect to local service [%s:%d] error: %v", localInfo.LocalIP, localInfo.LocalPort, err)

View File

@ -17,7 +17,6 @@ package client
import ( import (
"context" "context"
"crypto/tls" "crypto/tls"
"errors"
"fmt" "fmt"
"io" "io"
"net" "net"
@ -34,6 +33,7 @@ import (
"github.com/fatedier/frp/pkg/transport" "github.com/fatedier/frp/pkg/transport"
"github.com/fatedier/frp/pkg/util/log" "github.com/fatedier/frp/pkg/util/log"
frpNet "github.com/fatedier/frp/pkg/util/net" frpNet "github.com/fatedier/frp/pkg/util/net"
"github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/version" "github.com/fatedier/frp/pkg/util/version"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
libdial "github.com/fatedier/golib/net/dial" libdial "github.com/fatedier/golib/net/dial"
@ -109,7 +109,7 @@ func (svr *Service) Run() error {
if svr.cfg.LoginFailExit { if svr.cfg.LoginFailExit {
return err return err
} }
time.Sleep(10 * time.Second) util.RandomSleep(10*time.Second, 0.9, 1.1)
} else { } else {
// login success // login success
ctl := NewControl(svr.ctx, svr.runID, conn, session, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.serverUDPPort, svr.authSetter) ctl := NewControl(svr.ctx, svr.runID, conn, session, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.serverUDPPort, svr.authSetter)
@ -158,8 +158,11 @@ func (svr *Service) keepControllerWorking() {
// the first three retry with no delay // the first three retry with no delay
if reconnectCounts > 3 { if reconnectCounts > 3 {
time.Sleep(reconnectDelay) util.RandomSleep(reconnectDelay, 0.9, 1.1)
xl.Info("wait %v to reconnect", reconnectDelay)
reconnectDelay *= 2 reconnectDelay *= 2
} else {
util.RandomSleep(time.Second, 0, 0.5)
} }
reconnectCounts++ reconnectCounts++
@ -175,18 +178,12 @@ func (svr *Service) keepControllerWorking() {
xl.Info("try to reconnect to server...") xl.Info("try to reconnect to server...")
conn, session, err := svr.login() conn, session, err := svr.login()
if err != nil { if err != nil {
xl.Warn("reconnect to server error: %v", err) xl.Warn("reconnect to server error: %v, wait %v for another retry", err, delayTime)
time.Sleep(delayTime) util.RandomSleep(delayTime, 0.9, 1.1)
opErr := &net.OpError{} delayTime = delayTime * 2
// quick retry for dial error if delayTime > maxDelayTime {
if errors.As(err, &opErr) && opErr.Op == "dial" { delayTime = maxDelayTime
delayTime = 2 * time.Second
} else {
delayTime = delayTime * 2
if delayTime > maxDelayTime {
delayTime = maxDelayTime
}
} }
continue continue
} }
@ -245,6 +242,7 @@ func (svr *Service) login() (conn net.Conn, session *fmux.Session, err error) {
} }
dialOptions = append(dialOptions, dialOptions = append(dialOptions,
libdial.WithProtocol(protocol), libdial.WithProtocol(protocol),
libdial.WithTimeout(time.Duration(svr.cfg.DialServerTimeout)*time.Second),
libdial.WithProxy(proxyType, addr), libdial.WithProxy(proxyType, addr),
libdial.WithProxyAuth(auth), libdial.WithProxyAuth(auth),
libdial.WithTLSConfig(tlsConfig), libdial.WithTLSConfig(tlsConfig),

View File

@ -6,6 +6,9 @@
server_addr = 0.0.0.0 server_addr = 0.0.0.0
server_port = 7000 server_port = 7000
# The maximum amount of time a dial to server will wait for a connect to complete. Default value is 10 seconds.
# dial_server_timeout = 10
# if you want to connect frps by http proxy or socks5 proxy or ntlm proxy, you can set http_proxy here or in global environment variables # if you want to connect frps by http proxy or socks5 proxy or ntlm proxy, you can set http_proxy here or in global environment variables
# it only works when protocol is tcp # it only works when protocol is tcp
# http_proxy = http://user:passwd@192.168.1.128:8080 # http_proxy = http://user:passwd@192.168.1.128:8080
@ -48,6 +51,12 @@ oidc_audience =
# It will be used to get an OIDC token if AuthenticationMethod == "oidc". By default, this value is "". # It will be used to get an OIDC token if AuthenticationMethod == "oidc". By default, this value is "".
oidc_token_endpoint_url = oidc_token_endpoint_url =
# oidc_additional_xxx specifies additional parameters to be sent to the OIDC Token Endpoint.
# For example, if you want to specify the "audience" parameter, you can set as follow.
# frp will add "audience=<value>" "var1=<value>" to the additional parameters.
# oidc_additional_audience = https://dev.auth.com/api/v2/
# oidc_additional_var1 = foobar
# set admin address for control frpc's action by http api such as reload # set admin address for control frpc's action by http api such as reload
admin_addr = 127.0.0.1 admin_addr = 127.0.0.1
admin_port = 7400 admin_port = 7400

View File

@ -70,7 +70,7 @@ The response can look like any of the following:
### Operation ### Operation
Currently `Login`, `NewProxy`, `Ping`, `NewWorkConn` and `NewUserConn` operations are supported. Currently `Login`, `NewProxy`, `CloseProxy`, `Ping`, `NewWorkConn` and `NewUserConn` operations are supported.
#### Login #### Login
@ -136,6 +136,26 @@ Create new proxy
} }
``` ```
#### CloseProxy
A previously created proxy is closed.
Please note that one request will be sent for every proxy that is closed, do **NOT** use this
if you have too many proxies bound to a single client, as this may exhaust the server's resources.
```
{
"content": {
"user": {
"user": <string>,
"metas": map<string>string
"run_id": <string>
},
"proxy_name": <string>
}
}
```
#### Ping #### Ping
Heartbeat from frpc Heartbeat from frpc

2
go.mod
View File

@ -6,7 +6,7 @@ require (
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
github.com/coreos/go-oidc v2.2.1+incompatible github.com/coreos/go-oidc v2.2.1+incompatible
github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb
github.com/fatedier/golib v0.1.1-0.20220119075718-78e5cf8c00ee github.com/fatedier/golib v0.1.1-0.20220218075713-264f72dfbfd9
github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible
github.com/go-playground/validator/v10 v10.6.1 github.com/go-playground/validator/v10 v10.6.1
github.com/google/uuid v1.2.0 github.com/google/uuid v1.2.0

6
go.sum
View File

@ -88,8 +88,10 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb h1:wCrNShQidLmvVWn/0PikGmpdP0vtQmnvyRg3ZBEhczw= github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb h1:wCrNShQidLmvVWn/0PikGmpdP0vtQmnvyRg3ZBEhczw=
github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb/go.mod h1:wx3gB6dbIfBRcucp94PI9Bt3I0F2c/MyNEWuhzpWiwk= github.com/fatedier/beego v0.0.0-20171024143340-6c6a4f5bd5eb/go.mod h1:wx3gB6dbIfBRcucp94PI9Bt3I0F2c/MyNEWuhzpWiwk=
github.com/fatedier/golib v0.1.1-0.20220119075718-78e5cf8c00ee h1:iS0wlj2uZPxh3pciAf/HTzi88Kqu7DPh1jNKgJaFhtI= github.com/fatedier/golib v0.1.1-0.20220218073251-9509a597216b h1:5r5/G3NFsFK+7svxvxZYA8yy8Ubs4hWIq+QYYMgEBe8=
github.com/fatedier/golib v0.1.1-0.20220119075718-78e5cf8c00ee/go.mod h1:fLV0TLwHqrnB/L3jbNl67Gn6PCLggDGHniX1wLrA2Qo= github.com/fatedier/golib v0.1.1-0.20220218073251-9509a597216b/go.mod h1:fLV0TLwHqrnB/L3jbNl67Gn6PCLggDGHniX1wLrA2Qo=
github.com/fatedier/golib v0.1.1-0.20220218075713-264f72dfbfd9 h1:AOGf9Z1ri+3MiyGIAYXe+shEXx6/uVGJlufb6ZfnZls=
github.com/fatedier/golib v0.1.1-0.20220218075713-264f72dfbfd9/go.mod h1:fLV0TLwHqrnB/L3jbNl67Gn6PCLggDGHniX1wLrA2Qo=
github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible h1:ssXat9YXFvigNge/IkkZvFMn8yeYKFX+uI6wn2mLJ74= github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible h1:ssXat9YXFvigNge/IkkZvFMn8yeYKFX+uI6wn2mLJ74=
github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible/go.mod h1:YpCOaxj7vvMThhIQ9AfTOPW2sfztQR5WDfs7AflSy4s= github.com/fatedier/kcp-go v2.0.4-0.20190803094908-fe8645b0a904+incompatible/go.mod h1:YpCOaxj7vvMThhIQ9AfTOPW2sfztQR5WDfs7AflSy4s=
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=

View File

@ -40,14 +40,20 @@ type OidcClientConfig struct {
// It will be used to get an OIDC token if AuthenticationMethod == "oidc". // It will be used to get an OIDC token if AuthenticationMethod == "oidc".
// By default, this value is "". // By default, this value is "".
OidcTokenEndpointURL string `ini:"oidc_token_endpoint_url" json:"oidc_token_endpoint_url"` OidcTokenEndpointURL string `ini:"oidc_token_endpoint_url" json:"oidc_token_endpoint_url"`
// OidcAdditionalEndpointParams specifies additional parameters to be sent
// this field will be transfer to map[string][]string in OIDC token generator
// The field will be set by prefix "oidc_additional_"
OidcAdditionalEndpointParams map[string]string `ini:"-" json:"oidc_additional_endpoint_params"`
} }
func getDefaultOidcClientConf() OidcClientConfig { func getDefaultOidcClientConf() OidcClientConfig {
return OidcClientConfig{ return OidcClientConfig{
OidcClientID: "", OidcClientID: "",
OidcClientSecret: "", OidcClientSecret: "",
OidcAudience: "", OidcAudience: "",
OidcTokenEndpointURL: "", OidcTokenEndpointURL: "",
OidcAdditionalEndpointParams: make(map[string]string),
} }
} }
@ -88,11 +94,17 @@ type OidcAuthProvider struct {
} }
func NewOidcAuthSetter(baseCfg BaseConfig, cfg OidcClientConfig) *OidcAuthProvider { func NewOidcAuthSetter(baseCfg BaseConfig, cfg OidcClientConfig) *OidcAuthProvider {
eps := make(map[string][]string)
for k, v := range cfg.OidcAdditionalEndpointParams {
eps[k] = []string{v}
}
tokenGenerator := &clientcredentials.Config{ tokenGenerator := &clientcredentials.Config{
ClientID: cfg.OidcClientID, ClientID: cfg.OidcClientID,
ClientSecret: cfg.OidcClientSecret, ClientSecret: cfg.OidcClientSecret,
Scopes: []string{cfg.OidcAudience}, Scopes: []string{cfg.OidcAudience},
TokenURL: cfg.OidcTokenEndpointURL, TokenURL: cfg.OidcTokenEndpointURL,
EndpointParams: eps,
} }
return &OidcAuthProvider{ return &OidcAuthProvider{

View File

@ -38,6 +38,8 @@ type ClientCommonConf struct {
// ServerPort specifies the port to connect to the server on. By default, // ServerPort specifies the port to connect to the server on. By default,
// this value is 7000. // this value is 7000.
ServerPort int `ini:"server_port" json:"server_port"` ServerPort int `ini:"server_port" json:"server_port"`
// The maximum amount of time a dial to server will wait for a connect to complete.
DialServerTimeout int64 `ini:"dial_server_timeout" json:"dial_server_timeout"`
// ConnectServerLocalIP specifies the address of the client bind when it connect to server. // ConnectServerLocalIP specifies the address of the client bind when it connect to server.
// By default, this value is empty. // By default, this value is empty.
// this value only use in TCP/Websocket protocol. Not support in KCP protocol. // this value only use in TCP/Websocket protocol. Not support in KCP protocol.
@ -157,6 +159,7 @@ func GetDefaultClientConf() ClientCommonConf {
ClientConfig: auth.GetDefaultClientConf(), ClientConfig: auth.GetDefaultClientConf(),
ServerAddr: "0.0.0.0", ServerAddr: "0.0.0.0",
ServerPort: 7000, ServerPort: 7000,
DialServerTimeout: 10,
HTTPProxy: os.Getenv("http_proxy"), HTTPProxy: os.Getenv("http_proxy"),
LogFile: "console", LogFile: "console",
LogWay: "console", LogWay: "console",
@ -258,6 +261,8 @@ func UnmarshalClientConfFromIni(source interface{}) (ClientCommonConf, error) {
} }
common.Metas = GetMapWithoutPrefix(s.KeysHash(), "meta_") common.Metas = GetMapWithoutPrefix(s.KeysHash(), "meta_")
common.ClientConfig.OidcAdditionalEndpointParams = GetMapWithoutPrefix(s.KeysHash(), "oidc_additional_")
return common, nil return common, nil
} }

View File

@ -261,6 +261,7 @@ func Test_LoadClientCommonConf(t *testing.T) {
}, },
ServerAddr: "0.0.0.9", ServerAddr: "0.0.0.9",
ServerPort: 7009, ServerPort: 7009,
DialServerTimeout: 10,
HTTPProxy: "http://user:passwd@192.168.1.128:8080", HTTPProxy: "http://user:passwd@192.168.1.128:8080",
LogFile: "./frpc.log9", LogFile: "./frpc.log9",
LogWay: "file", LogWay: "file",

View File

@ -18,6 +18,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"strings"
"github.com/fatedier/frp/pkg/util/util" "github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/xlog" "github.com/fatedier/frp/pkg/util/xlog"
@ -26,6 +27,7 @@ import (
type Manager struct { type Manager struct {
loginPlugins []Plugin loginPlugins []Plugin
newProxyPlugins []Plugin newProxyPlugins []Plugin
closeProxyPlugins []Plugin
pingPlugins []Plugin pingPlugins []Plugin
newWorkConnPlugins []Plugin newWorkConnPlugins []Plugin
newUserConnPlugins []Plugin newUserConnPlugins []Plugin
@ -35,6 +37,7 @@ func NewManager() *Manager {
return &Manager{ return &Manager{
loginPlugins: make([]Plugin, 0), loginPlugins: make([]Plugin, 0),
newProxyPlugins: make([]Plugin, 0), newProxyPlugins: make([]Plugin, 0),
closeProxyPlugins: make([]Plugin, 0),
pingPlugins: make([]Plugin, 0), pingPlugins: make([]Plugin, 0),
newWorkConnPlugins: make([]Plugin, 0), newWorkConnPlugins: make([]Plugin, 0),
newUserConnPlugins: make([]Plugin, 0), newUserConnPlugins: make([]Plugin, 0),
@ -48,6 +51,9 @@ func (m *Manager) Register(p Plugin) {
if p.IsSupport(OpNewProxy) { if p.IsSupport(OpNewProxy) {
m.newProxyPlugins = append(m.newProxyPlugins, p) m.newProxyPlugins = append(m.newProxyPlugins, p)
} }
if p.IsSupport(OpCloseProxy) {
m.closeProxyPlugins = append(m.closeProxyPlugins, p)
}
if p.IsSupport(OpPing) { if p.IsSupport(OpPing) {
m.pingPlugins = append(m.pingPlugins, p) m.pingPlugins = append(m.pingPlugins, p)
} }
@ -127,6 +133,32 @@ func (m *Manager) NewProxy(content *NewProxyContent) (*NewProxyContent, error) {
return content, nil return content, nil
} }
func (m *Manager) CloseProxy(content *CloseProxyContent) error {
if len(m.closeProxyPlugins) == 0 {
return nil
}
errs := make([]string, 0)
reqid, _ := util.RandID()
xl := xlog.New().AppendPrefix("reqid: " + reqid)
ctx := xlog.NewContext(context.Background(), xl)
ctx = NewReqidContext(ctx, reqid)
for _, p := range m.closeProxyPlugins {
_, _, err := p.Handle(ctx, OpCloseProxy, *content)
if err != nil {
xl.Warn("send CloseProxy request to plugin [%s] error: %v", p.Name(), err)
errs = append(errs, fmt.Sprintf("[%s]: %v", p.Name(), err))
}
}
if len(errs) > 0 {
return fmt.Errorf("send CloseProxy request to plugin errors: %s", strings.Join(errs, "; "))
} else {
return nil
}
}
func (m *Manager) Ping(content *PingContent) (*PingContent, error) { func (m *Manager) Ping(content *PingContent) (*PingContent, error) {
if len(m.pingPlugins) == 0 { if len(m.pingPlugins) == 0 {
return content, nil return content, nil

View File

@ -23,6 +23,7 @@ const (
OpLogin = "Login" OpLogin = "Login"
OpNewProxy = "NewProxy" OpNewProxy = "NewProxy"
OpCloseProxy = "CloseProxy"
OpPing = "Ping" OpPing = "Ping"
OpNewWorkConn = "NewWorkConn" OpNewWorkConn = "NewWorkConn"
OpNewUserConn = "NewUserConn" OpNewUserConn = "NewUserConn"

View File

@ -48,6 +48,11 @@ type NewProxyContent struct {
msg.NewProxy msg.NewProxy
} }
type CloseProxyContent struct {
User UserInfo `json:"user"`
msg.CloseProxy
}
type PingContent struct { type PingContent struct {
User UserInfo `json:"user"` User UserInfo `json:"user"`
msg.Ping msg.Ping

View File

@ -100,6 +100,8 @@ func NewClientTLSConfig(certPath, keyPath, caPath, serverName string) (*tls.Conf
base.Certificates = []tls.Certificate{*cert} base.Certificates = []tls.Certificate{*cert}
} }
base.ServerName = serverName
if caPath != "" { if caPath != "" {
pool, err := newCertPool(caPath) pool, err := newCertPool(caPath)
if err != nil { if err != nil {
@ -107,7 +109,6 @@ func NewClientTLSConfig(certPath, keyPath, caPath, serverName string) (*tls.Conf
} }
base.RootCAs = pool base.RootCAs = pool
base.ServerName = serverName
base.InsecureSkipVerify = false base.InsecureSkipVerify = false
} else { } else {
base.InsecureSkipVerify = true base.InsecureSkipVerify = true

View File

@ -19,9 +19,11 @@ import (
"crypto/rand" "crypto/rand"
"encoding/hex" "encoding/hex"
"fmt" "fmt"
mathrand "math/rand"
"net" "net"
"strconv" "strconv"
"strings" "strings"
"time"
) )
// RandID return a rand string used in frp. // RandID return a rand string used in frp.
@ -109,3 +111,17 @@ func GenerateResponseErrorString(summary string, err error, detailed bool) strin
} }
return summary return summary
} }
func RandomSleep(duration time.Duration, minRatio, maxRatio float64) time.Duration {
min := int64(minRatio * 1000.0)
max := int64(maxRatio * 1000.0)
var n int64
if max <= min {
n = min
} else {
n = mathrand.Int63n(max-min) + min
}
d := duration * time.Duration(n) / time.Duration(1000)
time.Sleep(d)
return d
}

View File

@ -19,7 +19,7 @@ import (
"strings" "strings"
) )
var version string = "0.39.1" var version string = "0.40.0"
func Full() string { func Full() string {
return version return version

View File

@ -376,6 +376,20 @@ func (ctl *Control) stoper() {
pxy.Close() pxy.Close()
ctl.pxyManager.Del(pxy.GetName()) ctl.pxyManager.Del(pxy.GetName())
metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConf().GetBaseInfo().ProxyType) metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConf().GetBaseInfo().ProxyType)
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)
}()
} }
ctl.allShutdown.Done() ctl.allShutdown.Done()
@ -564,5 +578,20 @@ func (ctl *Control) CloseProxy(closeMsg *msg.CloseProxy) (err error) {
ctl.mu.Unlock() ctl.mu.Unlock()
metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConf().GetBaseInfo().ProxyType) metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConf().GetBaseInfo().ProxyType)
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

@ -268,5 +268,4 @@ var _ = Describe("[Feature: Client-Server]", func() {
}) })
} }
}) })
}) })

View File

@ -12,7 +12,7 @@ import (
// RunProcesses run multiple processes from templates. // RunProcesses run multiple processes from templates.
// The first template should always be frps. // The first template should always be frps.
func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []string) { func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []string) ([]*process.Process, []*process.Process) {
templates := make([]string, 0, len(serverTemplates)+len(clientTemplates)) templates := make([]string, 0, len(serverTemplates)+len(clientTemplates))
for _, t := range serverTemplates { for _, t := range serverTemplates {
templates = append(templates, t) templates = append(templates, t)
@ -28,6 +28,7 @@ func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []str
f.usedPorts[name] = port f.usedPorts[name] = port
} }
currentServerProcesses := make([]*process.Process, 0, len(serverTemplates))
for i := range serverTemplates { for i := range serverTemplates {
path := filepath.Join(f.TempDirectory, fmt.Sprintf("frp-e2e-server-%d", i)) path := filepath.Join(f.TempDirectory, fmt.Sprintf("frp-e2e-server-%d", i))
err = os.WriteFile(path, []byte(outs[i]), 0666) err = os.WriteFile(path, []byte(outs[i]), 0666)
@ -37,11 +38,13 @@ func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []str
p := process.NewWithEnvs(TestContext.FRPServerPath, []string{"-c", path}, f.osEnvs) p := process.NewWithEnvs(TestContext.FRPServerPath, []string{"-c", path}, f.osEnvs)
f.serverConfPaths = append(f.serverConfPaths, path) f.serverConfPaths = append(f.serverConfPaths, path)
f.serverProcesses = append(f.serverProcesses, p) f.serverProcesses = append(f.serverProcesses, p)
currentServerProcesses = append(currentServerProcesses, p)
err = p.Start() err = p.Start()
ExpectNoError(err) ExpectNoError(err)
} }
time.Sleep(time.Second) time.Sleep(time.Second)
currentClientProcesses := make([]*process.Process, 0, len(clientTemplates))
for i := range clientTemplates { for i := range clientTemplates {
index := i + len(serverTemplates) index := i + len(serverTemplates)
path := filepath.Join(f.TempDirectory, fmt.Sprintf("frp-e2e-client-%d", i)) path := filepath.Join(f.TempDirectory, fmt.Sprintf("frp-e2e-client-%d", i))
@ -52,11 +55,14 @@ func (f *Framework) RunProcesses(serverTemplates []string, clientTemplates []str
p := process.NewWithEnvs(TestContext.FRPClientPath, []string{"-c", path}, f.osEnvs) p := process.NewWithEnvs(TestContext.FRPClientPath, []string{"-c", path}, f.osEnvs)
f.clientConfPaths = append(f.clientConfPaths, path) f.clientConfPaths = append(f.clientConfPaths, path)
f.clientProcesses = append(f.clientProcesses, p) f.clientProcesses = append(f.clientProcesses, p)
currentClientProcesses = append(currentClientProcesses, p)
err = p.Start() err = p.Start()
ExpectNoError(err) ExpectNoError(err)
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)
} }
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)
return currentServerProcesses, currentClientProcesses
} }
func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) { func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) {

View File

@ -158,6 +158,56 @@ var _ = Describe("[Feature: Server-Plugins]", func() {
}) })
}) })
Describe("CloseProxy", func() {
newFunc := func() *plugin.Request {
var r plugin.Request
r.Content = &plugin.CloseProxyContent{}
return &r
}
It("Validate Info", func() {
localPort := f.AllocPort()
var recordProxyName string
handler := func(req *plugin.Request) *plugin.Response {
var ret plugin.Response
content := req.Content.(*plugin.CloseProxyContent)
recordProxyName = content.ProxyName
return &ret
}
pluginServer := NewHTTPPluginServer(localPort, newFunc, handler, nil)
f.RunServer("", pluginServer)
serverConf := consts.DefaultServerConfig + fmt.Sprintf(`
[plugin.test]
addr = 127.0.0.1:%d
path = /handler
ops = CloseProxy
`, localPort)
clientConf := consts.DefaultClientConfig
remotePort := f.AllocPort()
clientConf += fmt.Sprintf(`
[tcp]
type = tcp
local_port = {{ .%s }}
remote_port = %d
`, framework.TCPEchoServerPort, remotePort)
_, clients := f.RunProcesses([]string{serverConf}, []string{clientConf})
framework.NewRequestExpect(f).Port(remotePort).Ensure()
for _, c := range clients {
c.Stop()
}
time.Sleep(1 * time.Second)
framework.ExpectEqual(recordProxyName, "tcp")
})
})
Describe("Ping", func() { Describe("Ping", func() {
newFunc := func() *plugin.Request { newFunc := func() *plugin.Request {
var r plugin.Request var r plugin.Request