frp/server/service.go

656 lines
19 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 (
2018-08-10 11:43:08 +08:00
"bytes"
2019-10-12 20:13:12 +08:00
"context"
2019-03-11 14:14:31 +08:00
"crypto/tls"
2017-03-09 02:03:47 +08:00
"fmt"
"io"
2017-12-13 03:27:43 +08:00
"net"
"net/http"
"os"
2021-03-10 20:19:58 +08:00
"strconv"
2017-03-09 02:03:47 +08:00
"time"
2022-08-29 01:02:53 +08:00
"github.com/fatedier/golib/net/mux"
fmux "github.com/hashicorp/yamux"
quic "github.com/quic-go/quic-go"
"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"
v1 "github.com/fatedier/frp/pkg/config/v1"
2020-09-23 13:49:14 +08:00
modelmetrics "github.com/fatedier/frp/pkg/metrics"
"github.com/fatedier/frp/pkg/msg"
"github.com/fatedier/frp/pkg/nathole"
plugin "github.com/fatedier/frp/pkg/plugin/server"
2023-11-21 11:19:35 +08:00
"github.com/fatedier/frp/pkg/ssh"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/transport"
2023-11-27 15:47:49 +08:00
httppkg "github.com/fatedier/frp/pkg/util/http"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/util/log"
2023-11-27 15:47:49 +08:00
netpkg "github.com/fatedier/frp/pkg/util/net"
2020-09-23 13:49:14 +08:00
"github.com/fatedier/frp/pkg/util/tcpmux"
"github.com/fatedier/frp/pkg/util/util"
"github.com/fatedier/frp/pkg/util/version"
"github.com/fatedier/frp/pkg/util/vhost"
"github.com/fatedier/frp/pkg/util/xlog"
2019-01-15 00:11:08 +08:00
"github.com/fatedier/frp/server/controller"
2018-05-23 14:39:12 +08:00
"github.com/fatedier/frp/server/group"
"github.com/fatedier/frp/server/metrics"
2018-05-23 14:39:12 +08:00
"github.com/fatedier/frp/server/ports"
2019-01-15 00:11:08 +08:00
"github.com/fatedier/frp/server/proxy"
2020-05-24 11:28:57 +08:00
"github.com/fatedier/frp/server/visitor"
2017-03-09 02:03:47 +08:00
)
2017-05-10 00:46:42 +08:00
const (
connReadTimeout time.Duration = 10 * time.Second
vhostReadWriteTimeout time.Duration = 30 * time.Second
2017-05-10 00:46:42 +08:00
)
func init() {
// Disable quic-go's receive buffer warning.
os.Setenv("QUIC_GO_DISABLE_RECEIVE_BUFFER_WARNING", "true")
// Disable quic-go's ECN support by default. It may cause issues on certain operating systems.
if os.Getenv("QUIC_GO_DISABLE_ECN") == "" {
os.Setenv("QUIC_GO_DISABLE_ECN", "true")
}
}
// Server service
2017-03-09 02:03:47 +08:00
type Service struct {
// Dispatch connections to different handlers listen on same port
muxer *mux.Mux
// Accept connections from client
2019-10-12 20:13:12 +08:00
listener net.Listener
2017-03-09 02:03:47 +08:00
// Accept connections using kcp
2019-10-12 20:13:12 +08:00
kcpListener net.Listener
2017-06-04 19:56:21 +08:00
// Accept connections using quic
2023-07-21 14:34:44 +08:00
quicListener *quic.Listener
2018-08-03 19:41:54 +08:00
// Accept connections using websocket
2019-10-12 20:13:12 +08:00
websocketListener net.Listener
2018-08-03 19:41:54 +08:00
2019-03-11 14:14:31 +08:00
// Accept frp tls connections
2019-10-12 20:13:12 +08:00
tlsListener net.Listener
2019-03-11 14:14:31 +08:00
2023-11-27 15:47:49 +08:00
// Accept pipe connections from ssh tunnel gateway
sshTunnelListener *netpkg.InternalListener
2023-11-21 11:19:35 +08:00
2019-01-15 00:11:08 +08:00
// Manage all controllers
ctlManager *ControlManager
// Manage all proxies
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
// Manage all plugins
pluginManager *plugin.Manager
2019-07-31 00:41:58 +08:00
// HTTP vhost router
2020-05-24 17:48:37 +08:00
httpVhostRouter *vhost.Routers
2019-07-31 00:41:58 +08:00
2019-01-10 20:53:06 +08:00
// All resource managers and controllers
2019-01-15 00:11:08 +08:00
rc *controller.ResourceController
2023-11-27 15:47:49 +08:00
// web server for dashboard UI and apis
webServer *httppkg.Server
2023-11-21 11:19:35 +08:00
sshTunnelGateway *ssh.Gateway
// Verifies authentication based on selected method
authVerifier auth.Verifier
2019-03-11 14:14:31 +08:00
tlsConfig *tls.Config
cfg *v1.ServerConfig
// service context
ctx context.Context
// call cancel to stop service
cancel context.CancelFunc
2017-03-09 02:03:47 +08:00
}
2023-11-27 15:47:49 +08:00
func NewService(cfg *v1.ServerConfig) (*Service, error) {
tlsConfig, err := transport.NewServerTLSConfig(
2023-09-13 16:32:39 +08:00
cfg.Transport.TLS.CertFile,
cfg.Transport.TLS.KeyFile,
cfg.Transport.TLS.TrustedCaFile)
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, err
}
var webServer *httppkg.Server
if cfg.WebServer.Port > 0 {
ws, err := httppkg.NewServer(cfg.WebServer)
if err != nil {
return nil, err
}
webServer = ws
modelmetrics.EnableMem()
if cfg.EnablePrometheus {
modelmetrics.EnablePrometheus()
}
}
2023-11-27 15:47:49 +08:00
svr := &Service{
2019-12-20 20:28:28 +08:00
ctlManager: NewControlManager(),
2020-05-24 17:48:37 +08:00
pxyManager: proxy.NewManager(),
2019-12-20 20:28:28 +08:00
pluginManager: plugin.NewManager(),
2019-01-15 00:11:08 +08:00
rc: &controller.ResourceController{
2020-05-24 17:48:37 +08:00
VisitorManager: visitor.NewManager(),
TCPPortManager: ports.NewManager("tcp", cfg.ProxyBindAddr, cfg.AllowPorts),
UDPPortManager: ports.NewManager("udp", cfg.ProxyBindAddr, cfg.AllowPorts),
2019-01-10 20:53:06 +08:00
},
2023-11-27 15:47:49 +08:00
sshTunnelListener: netpkg.NewInternalListener(),
httpVhostRouter: vhost.NewRouters(),
authVerifier: auth.NewAuthVerifier(cfg.Auth),
webServer: webServer,
tlsConfig: tlsConfig,
cfg: cfg,
ctx: context.Background(),
}
if webServer != nil {
webServer.RouteRegister(svr.registerRouteHandlers)
2017-03-09 02:03:47 +08:00
}
2019-01-15 00:11:08 +08:00
2020-04-20 13:35:47 +08:00
// Create tcpmux httpconnect multiplexer.
2020-05-24 17:48:37 +08:00
if cfg.TCPMuxHTTPConnectPort > 0 {
2020-04-20 13:35:47 +08:00
var l net.Listener
address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.TCPMuxHTTPConnectPort))
l, err = net.Listen("tcp", address)
2020-04-20 13:35:47 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create server listener error, %v", err)
2020-04-20 13:35:47 +08:00
}
svr.rc.TCPMuxHTTPConnectMuxer, err = tcpmux.NewHTTPConnectTCPMuxer(l, cfg.TCPMuxPassthrough, vhostReadWriteTimeout)
2020-04-20 13:35:47 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create vhost tcpMuxer error, %v", err)
2020-04-20 13:35:47 +08:00
}
2024-03-12 13:58:53 +08:00
log.Infof("tcpmux httpconnect multiplexer listen on %s, passthough: %v", address, cfg.TCPMuxPassthrough)
2020-04-20 13:35:47 +08:00
}
2019-12-20 20:28:28 +08:00
// Init all plugins
for _, p := range cfg.HTTPPlugins {
svr.pluginManager.Register(plugin.NewHTTPPluginOptions(p))
2024-03-12 13:58:53 +08:00
log.Infof("plugin [%s] has been registered", p.Name)
2019-12-20 20:28:28 +08:00
}
svr.rc.PluginManager = svr.pluginManager
2019-12-20 20:28:28 +08:00
2019-01-15 00:11:08 +08:00
// Init group controller
2020-05-24 17:48:37 +08:00
svr.rc.TCPGroupCtl = group.NewTCPGroupCtl(svr.rc.TCPPortManager)
2017-03-09 02:03:47 +08:00
2019-07-31 00:41:58 +08:00
// Init HTTP group controller
svr.rc.HTTPGroupCtl = group.NewHTTPGroupController(svr.httpVhostRouter)
2020-04-20 13:35:47 +08:00
// Init TCP mux group controller
2020-05-24 17:48:37 +08:00
svr.rc.TCPMuxGroupCtl = group.NewTCPMuxGroupCtl(svr.rc.TCPMuxHTTPConnectMuxer)
2020-04-20 13:35:47 +08:00
2019-04-25 12:29:34 +08:00
// Init 404 not found page
vhost.NotFoundPagePath = cfg.Custom404Page
var (
httpMuxOn bool
httpsMuxOn bool
)
if cfg.BindAddr == cfg.ProxyBindAddr {
2020-05-24 17:48:37 +08:00
if cfg.BindPort == cfg.VhostHTTPPort {
httpMuxOn = true
}
2020-05-24 17:48:37 +08:00
if cfg.BindPort == cfg.VhostHTTPSPort {
httpsMuxOn = true
}
}
2017-03-09 02:03:47 +08:00
// Listen for accepting connections from client.
2021-03-10 20:19:58 +08:00
address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.BindPort))
ln, err := net.Listen("tcp", address)
2017-03-09 02:03:47 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create server listener error, %v", err)
2017-03-09 02:03:47 +08:00
}
2018-08-03 19:41:54 +08:00
2018-08-10 12:02:38 +08:00
svr.muxer = mux.NewMux(ln)
svr.muxer.SetKeepAlive(time.Duration(cfg.Transport.TCPKeepAlive) * time.Second)
2022-08-29 01:02:53 +08:00
go func() {
_ = svr.muxer.Serve()
}()
2018-08-03 19:41:54 +08:00
ln = svr.muxer.DefaultListener()
2019-10-12 20:13:12 +08:00
svr.listener = ln
2024-03-12 13:58:53 +08:00
log.Infof("frps tcp listen on %s", address)
2017-06-04 19:56:21 +08:00
// Listen for accepting connections from client using kcp protocol.
2020-05-24 17:48:37 +08:00
if cfg.KCPBindPort > 0 {
2021-03-10 20:19:58 +08:00
address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.KCPBindPort))
2023-11-27 15:47:49 +08:00
svr.kcpListener, err = netpkg.ListenKcp(address)
2017-06-04 19:56:21 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("listen on kcp udp address %s error: %v", address, err)
2017-06-04 19:56:21 +08:00
}
2024-03-12 13:58:53 +08:00
log.Infof("frps kcp listen on udp %s", address)
2017-06-04 19:56:21 +08:00
}
2017-03-09 02:03:47 +08:00
if cfg.QUICBindPort > 0 {
address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.QUICBindPort))
quicTLSCfg := tlsConfig.Clone()
quicTLSCfg.NextProtos = []string{"frp"}
2022-12-18 18:43:42 +08:00
svr.quicListener, err = quic.ListenAddr(address, quicTLSCfg, &quic.Config{
MaxIdleTimeout: time.Duration(cfg.Transport.QUIC.MaxIdleTimeout) * time.Second,
MaxIncomingStreams: int64(cfg.Transport.QUIC.MaxIncomingStreams),
KeepAlivePeriod: time.Duration(cfg.Transport.QUIC.KeepalivePeriod) * time.Second,
2022-12-18 18:43:42 +08:00
})
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("listen on quic udp address %s error: %v", address, err)
}
2024-03-12 13:58:53 +08:00
log.Infof("frps quic listen on %s", address)
2023-11-21 11:19:35 +08:00
}
if cfg.SSHTunnelGateway.BindPort > 0 {
2023-11-27 15:47:49 +08:00
sshGateway, err := ssh.NewGateway(cfg.SSHTunnelGateway, cfg.ProxyBindAddr, svr.sshTunnelListener)
2023-11-21 11:19:35 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create ssh gateway error: %v", err)
2023-11-21 11:19:35 +08:00
}
svr.sshTunnelGateway = sshGateway
2024-03-12 13:58:53 +08:00
log.Infof("frps sshTunnelGateway listen on port %d", cfg.SSHTunnelGateway.BindPort)
}
2018-08-10 11:43:08 +08:00
// Listen for accepting connections from client using websocket protocol.
2023-11-27 15:47:49 +08:00
websocketPrefix := []byte("GET " + netpkg.FrpWebsocketPath)
2018-08-10 11:43:08 +08:00
websocketLn := svr.muxer.Listen(0, uint32(len(websocketPrefix)), func(data []byte) bool {
return bytes.Equal(data, websocketPrefix)
})
2023-11-27 15:47:49 +08:00
svr.websocketListener = netpkg.NewWebsocketListener(websocketLn)
2018-08-10 11:43:08 +08:00
2017-03-09 02:03:47 +08:00
// Create http vhost muxer.
2020-05-24 17:48:37 +08:00
if cfg.VhostHTTPPort > 0 {
rp := vhost.NewHTTPReverseProxy(vhost.HTTPReverseProxyOptions{
ResponseHeaderTimeoutS: cfg.VhostHTTPTimeout,
2019-07-31 00:41:58 +08:00
}, svr.httpVhostRouter)
2020-05-24 17:48:37 +08:00
svr.rc.HTTPReverseProxy = rp
2017-12-13 03:27:43 +08:00
2021-03-10 20:19:58 +08:00
address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.VhostHTTPPort))
2017-12-13 03:27:43 +08:00
server := &http.Server{
Addr: address,
Handler: rp,
2017-03-09 02:03:47 +08:00
}
2017-12-13 03:27:43 +08:00
var l net.Listener
if httpMuxOn {
2018-08-10 11:43:08 +08:00
l = svr.muxer.ListenHttp(1)
} else {
l, err = net.Listen("tcp", address)
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create vhost http listener error, %v", err)
}
2017-03-09 02:03:47 +08:00
}
2022-08-29 01:02:53 +08:00
go func() {
_ = server.Serve(l)
}()
2024-03-12 13:58:53 +08:00
log.Infof("http service listen on %s", address)
2017-03-09 02:03:47 +08:00
}
// Create https vhost muxer.
2020-05-24 17:48:37 +08:00
if cfg.VhostHTTPSPort > 0 {
var l net.Listener
if httpsMuxOn {
2018-08-10 11:43:08 +08:00
l = svr.muxer.ListenHttps(1)
} else {
2021-03-10 20:19:58 +08:00
address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.VhostHTTPSPort))
l, err = net.Listen("tcp", address)
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create server listener error, %v", err)
}
2024-03-12 13:58:53 +08:00
log.Infof("https service listen on %s", address)
2017-03-09 02:03:47 +08:00
}
2020-05-24 17:48:37 +08:00
svr.rc.VhostHTTPSMuxer, err = vhost.NewHTTPSMuxer(l, vhostReadWriteTimeout)
2017-03-09 02:03:47 +08:00
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create vhost httpsMuxer error, %v", err)
2017-03-09 02:03:47 +08:00
}
2017-10-24 18:20:07 +08:00
}
2019-03-11 14:14:31 +08:00
// frp tls listener
svr.tlsListener = svr.muxer.Listen(2, 1, func(data []byte) bool {
// tls first byte can be 0x16 only when vhost https port is not same with bind port
2023-11-27 15:47:49 +08:00
return int(data[0]) == netpkg.FRPTLSHeadByte || int(data[0]) == 0x16
2019-03-11 14:14:31 +08:00
})
2017-10-24 18:20:07 +08:00
// Create nat hole controller.
nc, err := nathole.NewController(time.Duration(cfg.NatHoleAnalysisDataReserveHours) * time.Hour)
if err != nil {
2023-11-27 15:47:49 +08:00
return nil, fmt.Errorf("create nat hole controller error, %v", err)
2017-03-09 02:03:47 +08:00
}
svr.rc.NatHoleController = nc
2023-11-27 15:47:49 +08:00
return svr, nil
2017-03-09 02:03:47 +08:00
}
func (svr *Service) Run(ctx context.Context) {
ctx, cancel := context.WithCancel(ctx)
svr.ctx = ctx
svr.cancel = cancel
2023-11-27 15:47:49 +08:00
// run dashboard web server.
if svr.webServer != nil {
go func() {
2024-03-12 13:58:53 +08:00
log.Infof("dashboard listen on %s", svr.webServer.Address())
2023-11-27 15:47:49 +08:00
if err := svr.webServer.Run(); err != nil {
2024-03-12 13:58:53 +08:00
log.Warnf("dashboard server exit with error: %v", err)
2023-11-27 15:47:49 +08:00
}
}()
}
go svr.HandleListener(svr.sshTunnelListener, true)
if svr.kcpListener != nil {
2023-11-21 11:19:35 +08:00
go svr.HandleListener(svr.kcpListener, false)
2017-06-04 19:56:21 +08:00
}
if svr.quicListener != nil {
go svr.HandleQUICListener(svr.quicListener)
}
2023-11-21 11:19:35 +08:00
go svr.HandleListener(svr.websocketListener, false)
go svr.HandleListener(svr.tlsListener, false)
2018-08-10 11:43:08 +08:00
if svr.rc.NatHoleController != nil {
go svr.rc.NatHoleController.CleanWorker(svr.ctx)
}
2023-11-21 11:19:35 +08:00
if svr.sshTunnelGateway != nil {
go svr.sshTunnelGateway.Run()
}
svr.HandleListener(svr.listener, false)
<-svr.ctx.Done()
// service context may not be canceled by svr.Close(), we should call it here to release resources
if svr.listener != nil {
svr.Close()
}
2017-06-04 19:56:21 +08:00
}
func (svr *Service) Close() error {
if svr.kcpListener != nil {
svr.kcpListener.Close()
svr.kcpListener = nil
}
if svr.quicListener != nil {
svr.quicListener.Close()
svr.quicListener = nil
}
if svr.websocketListener != nil {
svr.websocketListener.Close()
svr.websocketListener = nil
}
if svr.tlsListener != nil {
svr.tlsListener.Close()
svr.tlsConfig = nil
}
if svr.listener != nil {
svr.listener.Close()
svr.listener = nil
}
svr.ctlManager.Close()
if svr.cancel != nil {
svr.cancel()
}
return nil
}
2023-11-27 15:47:49 +08:00
func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, internal bool) {
xl := xlog.FromContextSafe(ctx)
var (
rawMsg msg.Message
err error
)
2022-08-29 01:02:53 +08:00
_ = conn.SetReadDeadline(time.Now().Add(connReadTimeout))
if rawMsg, err = msg.ReadMsg(conn); err != nil {
2024-03-12 13:58:53 +08:00
log.Tracef("Failed to read message: %v", err)
conn.Close()
return
}
2022-08-29 01:02:53 +08:00
_ = conn.SetReadDeadline(time.Time{})
switch m := rawMsg.(type) {
case *msg.Login:
// server plugin hook
content := &plugin.LoginContent{
Login: *m,
ClientAddress: conn.RemoteAddr().String(),
}
retContent, err := svr.pluginManager.Login(content)
if err == nil {
m = &retContent.Login
2023-11-27 15:47:49 +08:00
err = svr.RegisterControl(conn, m, internal)
}
// If login failed, send error message there.
// Otherwise send success message in control's work goroutine.
if err != nil {
2024-03-12 13:58:53 +08:00
xl.Warnf("register control error: %v", err)
2022-08-29 01:02:53 +08:00
_ = msg.WriteMsg(conn, &msg.LoginResp{
Version: version.Full(),
Error: util.GenerateResponseErrorString("register control error", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
})
conn.Close()
}
case *msg.NewWorkConn:
if err := svr.RegisterWorkConn(conn, m); err != nil {
conn.Close()
}
case *msg.NewVisitorConn:
if err = svr.RegisterVisitorConn(conn, m); err != nil {
2024-03-12 13:58:53 +08:00
xl.Warnf("register visitor conn error: %v", err)
2022-08-29 01:02:53 +08:00
_ = msg.WriteMsg(conn, &msg.NewVisitorConnResp{
ProxyName: m.ProxyName,
Error: util.GenerateResponseErrorString("register visitor conn error", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
})
conn.Close()
} else {
2022-08-29 01:02:53 +08:00
_ = msg.WriteMsg(conn, &msg.NewVisitorConnResp{
ProxyName: m.ProxyName,
Error: "",
})
}
default:
2024-03-12 13:58:53 +08:00
log.Warnf("Error message type for the new connection [%s]", conn.RemoteAddr().String())
conn.Close()
}
}
2023-11-27 15:47:49 +08:00
// HandleListener accepts connections from client and call handleConnection to handle them.
// If internal is true, it means that this listener is used for internal communication like ssh tunnel gateway.
// TODO(fatedier): Pass some parameters of listener/connection through context to avoid passing too many parameters.
2023-11-21 11:19:35 +08:00
func (svr *Service) HandleListener(l net.Listener, internal bool) {
2017-03-09 02:03:47 +08:00
// Listen for incoming connections from client.
for {
2017-06-04 19:56:21 +08:00
c, err := l.Accept()
2017-03-09 02:03:47 +08:00
if err != nil {
2024-03-12 13:58:53 +08:00
log.Warnf("Listener for incoming connections from client closed")
2017-03-09 02:03:47 +08:00
return
}
2019-10-12 20:13:12 +08:00
// inject xlog object into net.Conn context
xl := xlog.New()
ctx := context.Background()
2023-11-27 15:47:49 +08:00
c = netpkg.NewContextConn(xlog.NewContext(ctx, xl), c)
2023-11-27 15:47:49 +08:00
if !internal {
2024-03-12 13:58:53 +08:00
log.Tracef("start check TLS connection...")
2023-11-27 15:47:49 +08:00
originConn := c
forceTLS := svr.cfg.Transport.TLS.Force
var isTLS, custom bool
c, isTLS, custom, err = netpkg.CheckAndEnableTLSServerConnWithTimeout(c, svr.tlsConfig, forceTLS, connReadTimeout)
if err != nil {
2024-03-12 13:58:53 +08:00
log.Warnf("CheckAndEnableTLSServerConnWithTimeout error: %v", err)
2023-11-27 15:47:49 +08:00
originConn.Close()
continue
}
2024-03-12 13:58:53 +08:00
log.Tracef("check TLS connection success, isTLS: %v custom: %v internal: %v", isTLS, custom, internal)
}
2017-03-09 02:03:47 +08:00
// Start a new goroutine to handle connection.
go func(ctx context.Context, frpConn net.Conn) {
2023-11-21 11:19:35 +08:00
if lo.FromPtr(svr.cfg.Transport.TCPMux) && !internal {
2018-05-05 00:09:39 +08:00
fmuxCfg := fmux.DefaultConfig()
fmuxCfg.KeepAliveInterval = time.Duration(svr.cfg.Transport.TCPMuxKeepaliveInterval) * time.Second
fmuxCfg.LogOutput = io.Discard
fmuxCfg.MaxStreamWindowSize = 6 * 1024 * 1024
2018-05-05 00:09:39 +08:00
session, err := fmux.Server(frpConn, fmuxCfg)
2017-03-09 02:03:47 +08:00
if err != nil {
2024-03-12 13:58:53 +08:00
log.Warnf("Failed to create mux connection: %v", err)
2017-03-09 02:03:47 +08:00
frpConn.Close()
return
}
for {
stream, err := session.AcceptStream()
if err != nil {
2024-03-12 13:58:53 +08:00
log.Debugf("Accept new mux stream error: %v", err)
2017-05-18 01:51:32 +08:00
session.Close()
return
}
2023-11-27 15:47:49 +08:00
go svr.handleConnection(ctx, stream, internal)
2017-03-09 02:03:47 +08:00
}
} else {
2023-11-27 15:47:49 +08:00
svr.handleConnection(ctx, frpConn, internal)
2017-03-09 02:03:47 +08:00
}
}(ctx, c)
2017-03-09 02:03:47 +08:00
}
}
2023-07-21 14:34:44 +08:00
func (svr *Service) HandleQUICListener(l *quic.Listener) {
// Listen for incoming connections from client.
for {
c, err := l.Accept(context.Background())
if err != nil {
2024-03-12 13:58:53 +08:00
log.Warnf("QUICListener for incoming connections from client closed")
return
}
// Start a new goroutine to handle connection.
go func(ctx context.Context, frpConn quic.Connection) {
for {
stream, err := frpConn.AcceptStream(context.Background())
if err != nil {
2024-03-12 13:58:53 +08:00
log.Debugf("Accept new quic mux stream error: %v", err)
_ = frpConn.CloseWithError(0, "")
return
}
2023-11-27 15:47:49 +08:00
go svr.handleConnection(ctx, netpkg.QuicStreamToNetConn(stream, frpConn), false)
}
}(context.Background(), c)
}
}
2023-11-27 15:47:49 +08:00
func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, internal bool) error {
2020-05-24 17:48:37 +08:00
// If client's RunID is empty, it's a new client, we just create a new controller.
2019-10-12 20:13:12 +08:00
// Otherwise, we check if there is one controller has the same run id. If so, we release previous controller and start new one.
var err error
2020-05-24 17:48:37 +08:00
if loginMsg.RunID == "" {
loginMsg.RunID, err = util.RandID()
2019-10-12 20:13:12 +08:00
if err != nil {
return err
2019-10-12 20:13:12 +08:00
}
}
2023-11-27 15:47:49 +08:00
ctx := netpkg.NewContextFromConn(ctlConn)
2019-10-12 20:13:12 +08:00
xl := xlog.FromContextSafe(ctx)
2020-05-24 17:48:37 +08:00
xl.AppendPrefix(loginMsg.RunID)
2019-10-12 20:13:12 +08:00
ctx = xlog.NewContext(ctx, xl)
2024-03-12 13:58:53 +08:00
xl.Infof("client login info: ip [%s] version [%s] hostname [%s] os [%s] arch [%s]",
2017-03-09 02:03:47 +08:00
ctlConn.RemoteAddr().String(), loginMsg.Version, loginMsg.Hostname, loginMsg.Os, loginMsg.Arch)
// Check auth.
2023-11-27 15:47:49 +08:00
authVerifier := svr.authVerifier
if internal && loginMsg.ClientSpec.AlwaysAuthPass {
authVerifier = auth.AlwaysPassVerifier
}
if err := authVerifier.VerifyLogin(loginMsg); err != nil {
return err
2017-03-09 02:03:47 +08:00
}
2023-11-27 15:47:49 +08:00
// TODO(fatedier): use SessionContext
ctl, err := NewControl(ctx, svr.rc, svr.pxyManager, svr.pluginManager, authVerifier, ctlConn, !internal, loginMsg, svr.cfg)
if err != nil {
2024-03-12 13:58:53 +08:00
xl.Warnf("create new controller error: %v", err)
// don't return detailed errors to client
2023-11-22 14:30:22 +08:00
return fmt.Errorf("unexpected error when creating new controller")
}
2020-05-24 17:48:37 +08:00
if oldCtl := svr.ctlManager.Add(loginMsg.RunID, ctl); oldCtl != nil {
oldCtl.WaitClosed()
2017-03-09 02:03:47 +08:00
}
ctl.Start()
2017-03-23 02:01:25 +08:00
// for statistics
metrics.Server.NewClient()
2019-01-15 00:11:08 +08:00
go func() {
// block until control closed
ctl.WaitClosed()
2020-05-24 17:48:37 +08:00
svr.ctlManager.Del(loginMsg.RunID, ctl)
2019-01-15 00:11:08 +08:00
}()
return nil
2017-03-09 02:03:47 +08:00
}
// RegisterWorkConn register a new work connection to control and proxies need it.
func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn) error {
2023-11-27 15:47:49 +08:00
xl := netpkg.NewLogFromConn(workConn)
2020-05-24 17:48:37 +08:00
ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
2017-03-09 02:03:47 +08:00
if !exist {
2024-03-12 13:58:53 +08:00
xl.Warnf("No client control found for run id [%s]", newMsg.RunID)
2020-05-24 17:48:37 +08:00
return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
2017-03-09 02:03:47 +08:00
}
// server plugin hook
content := &plugin.NewWorkConnContent{
User: plugin.UserInfo{
User: ctl.loginMsg.User,
Metas: ctl.loginMsg.Metas,
2020-05-24 17:48:37 +08:00
RunID: ctl.loginMsg.RunID,
},
NewWorkConn: *newMsg,
}
retContent, err := svr.pluginManager.NewWorkConn(content)
if err == nil {
newMsg = &retContent.NewWorkConn
// Check auth.
2023-11-27 15:47:49 +08:00
err = ctl.authVerifier.VerifyNewWorkConn(newMsg)
}
if err != nil {
2024-03-12 13:58:53 +08:00
xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
2022-08-29 01:02:53 +08:00
_ = msg.WriteMsg(workConn, &msg.StartWorkConn{
Error: util.GenerateResponseErrorString("invalid NewWorkConn", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
})
2020-05-24 17:48:37 +08:00
return fmt.Errorf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
}
return ctl.RegisterWorkConn(workConn)
2017-03-09 02:03:47 +08:00
}
2019-10-12 20:13:12 +08:00
func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVisitorConn) error {
visitorUser := ""
// TODO(deprecation): Compatible with old versions, can be without runID, user is empty. In later versions, it will be mandatory to include runID.
// If runID is required, it is not compatible with versions prior to v0.50.0.
if newMsg.RunID != "" {
ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
if !exist {
return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
}
visitorUser = ctl.loginMsg.User
}
2019-01-10 20:53:06 +08:00
return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
newMsg.UseEncryption, newMsg.UseCompression, visitorUser)
2017-06-26 03:02:33 +08:00
}