service.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684
  1. // Copyright 2017 fatedier, fatedier@gmail.com
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package server
  15. import (
  16. "bytes"
  17. "context"
  18. "crypto/tls"
  19. "fmt"
  20. "io"
  21. "net"
  22. "net/http"
  23. "os"
  24. "strconv"
  25. "time"
  26. "github.com/fatedier/golib/crypto"
  27. "github.com/fatedier/golib/net/mux"
  28. fmux "github.com/hashicorp/yamux"
  29. quic "github.com/quic-go/quic-go"
  30. "github.com/samber/lo"
  31. "github.com/fatedier/frp/pkg/auth"
  32. v1 "github.com/fatedier/frp/pkg/config/v1"
  33. modelmetrics "github.com/fatedier/frp/pkg/metrics"
  34. "github.com/fatedier/frp/pkg/msg"
  35. "github.com/fatedier/frp/pkg/nathole"
  36. plugin "github.com/fatedier/frp/pkg/plugin/server"
  37. "github.com/fatedier/frp/pkg/ssh"
  38. "github.com/fatedier/frp/pkg/transport"
  39. httppkg "github.com/fatedier/frp/pkg/util/http"
  40. "github.com/fatedier/frp/pkg/util/log"
  41. netpkg "github.com/fatedier/frp/pkg/util/net"
  42. "github.com/fatedier/frp/pkg/util/tcpmux"
  43. "github.com/fatedier/frp/pkg/util/util"
  44. "github.com/fatedier/frp/pkg/util/version"
  45. "github.com/fatedier/frp/pkg/util/vhost"
  46. "github.com/fatedier/frp/pkg/util/xlog"
  47. "github.com/fatedier/frp/server/controller"
  48. "github.com/fatedier/frp/server/group"
  49. "github.com/fatedier/frp/server/metrics"
  50. "github.com/fatedier/frp/server/ports"
  51. "github.com/fatedier/frp/server/proxy"
  52. "github.com/fatedier/frp/server/visitor"
  53. )
  54. const (
  55. connReadTimeout time.Duration = 10 * time.Second
  56. vhostReadWriteTimeout time.Duration = 30 * time.Second
  57. )
  58. func init() {
  59. crypto.DefaultSalt = "frp"
  60. // Disable quic-go's receive buffer warning.
  61. os.Setenv("QUIC_GO_DISABLE_RECEIVE_BUFFER_WARNING", "true")
  62. // Disable quic-go's ECN support by default. It may cause issues on certain operating systems.
  63. if os.Getenv("QUIC_GO_DISABLE_ECN") == "" {
  64. os.Setenv("QUIC_GO_DISABLE_ECN", "true")
  65. }
  66. }
  67. // Server service
  68. type Service struct {
  69. // Dispatch connections to different handlers listen on same port
  70. muxer *mux.Mux
  71. // Accept connections from client
  72. muxListener net.Listener
  73. // Accept connections using kcp
  74. kcpListener net.Listener
  75. // Accept connections using quic
  76. quicListener *quic.Listener
  77. // Accept connections using websocket
  78. websocketListener net.Listener
  79. // Accept frp tls connections
  80. tlsListener net.Listener
  81. // Accept pipe connections from ssh tunnel gateway
  82. sshTunnelListener *netpkg.InternalListener
  83. // Manage all controllers
  84. ctlManager *ControlManager
  85. // Manage all proxies
  86. pxyManager *proxy.Manager
  87. // Manage all plugins
  88. pluginManager *plugin.Manager
  89. // HTTP vhost router
  90. httpVhostRouter *vhost.Routers
  91. // All resource managers and controllers
  92. rc *controller.ResourceController
  93. // web server for dashboard UI and apis
  94. webServer *httppkg.Server
  95. sshTunnelGateway *ssh.Gateway
  96. // Verifies authentication based on selected method
  97. authVerifier auth.Verifier
  98. tlsConfig *tls.Config
  99. cfg *v1.ServerConfig
  100. // service context
  101. ctx context.Context
  102. // call cancel to stop service
  103. cancel context.CancelFunc
  104. // Track listeners so they can be closed manually
  105. vhostHTTPSListener net.Listener
  106. tcpmuxHTTPConnectListener net.Listener
  107. tcpListener net.Listener
  108. }
  109. func NewService(cfg *v1.ServerConfig) (*Service, error) {
  110. tlsConfig, err := transport.NewServerTLSConfig(
  111. cfg.Transport.TLS.CertFile,
  112. cfg.Transport.TLS.KeyFile,
  113. cfg.Transport.TLS.TrustedCaFile)
  114. if err != nil {
  115. return nil, err
  116. }
  117. var webServer *httppkg.Server
  118. if cfg.WebServer.Port > 0 {
  119. ws, err := httppkg.NewServer(cfg.WebServer)
  120. if err != nil {
  121. return nil, err
  122. }
  123. webServer = ws
  124. modelmetrics.EnableMem()
  125. if cfg.EnablePrometheus {
  126. modelmetrics.EnablePrometheus()
  127. }
  128. }
  129. svr := &Service{
  130. ctlManager: NewControlManager(),
  131. pxyManager: proxy.NewManager(),
  132. pluginManager: plugin.NewManager(),
  133. rc: &controller.ResourceController{
  134. VisitorManager: visitor.NewManager(),
  135. TCPPortManager: ports.NewManager("tcp", cfg.ProxyBindAddr, cfg.AllowPorts),
  136. UDPPortManager: ports.NewManager("udp", cfg.ProxyBindAddr, cfg.AllowPorts),
  137. },
  138. sshTunnelListener: netpkg.NewInternalListener(),
  139. httpVhostRouter: vhost.NewRouters(),
  140. authVerifier: auth.NewAuthVerifier(cfg.Auth),
  141. webServer: webServer,
  142. tlsConfig: tlsConfig,
  143. cfg: cfg,
  144. ctx: context.Background(),
  145. }
  146. if webServer != nil {
  147. webServer.RouteRegister(svr.registerRouteHandlers)
  148. }
  149. // Create tcpmux httpconnect multiplexer.
  150. if cfg.TCPMuxHTTPConnectPort > 0 {
  151. var l net.Listener
  152. address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.TCPMuxHTTPConnectPort))
  153. l, err = net.Listen("tcp", address)
  154. if err != nil {
  155. return nil, fmt.Errorf("create server listener error, %v", err)
  156. }
  157. // Save listener so it can be closed in svr.Close()
  158. svr.tcpmuxHTTPConnectListener = l
  159. svr.rc.TCPMuxHTTPConnectMuxer, err = tcpmux.NewHTTPConnectTCPMuxer(l, cfg.TCPMuxPassthrough, vhostReadWriteTimeout)
  160. if err != nil {
  161. return nil, fmt.Errorf("create vhost tcpMuxer error, %v", err)
  162. }
  163. log.Infof("tcpmux httpconnect multiplexer listen on %s, passthough: %v", address, cfg.TCPMuxPassthrough)
  164. }
  165. // Init all plugins
  166. for _, p := range cfg.HTTPPlugins {
  167. svr.pluginManager.Register(plugin.NewHTTPPluginOptions(p))
  168. log.Infof("plugin [%s] has been registered", p.Name)
  169. }
  170. svr.rc.PluginManager = svr.pluginManager
  171. // Init group controller
  172. svr.rc.TCPGroupCtl = group.NewTCPGroupCtl(svr.rc.TCPPortManager)
  173. // Init HTTP group controller
  174. svr.rc.HTTPGroupCtl = group.NewHTTPGroupController(svr.httpVhostRouter)
  175. // Init TCP mux group controller
  176. svr.rc.TCPMuxGroupCtl = group.NewTCPMuxGroupCtl(svr.rc.TCPMuxHTTPConnectMuxer)
  177. // Init 404 not found page
  178. vhost.NotFoundPagePath = cfg.Custom404Page
  179. var (
  180. httpMuxOn bool
  181. httpsMuxOn bool
  182. )
  183. if cfg.BindAddr == cfg.ProxyBindAddr {
  184. if cfg.BindPort == cfg.VhostHTTPPort {
  185. httpMuxOn = true
  186. }
  187. if cfg.BindPort == cfg.VhostHTTPSPort {
  188. httpsMuxOn = true
  189. }
  190. }
  191. // Listen for accepting connections from client.
  192. address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.BindPort))
  193. ln, err := net.Listen("tcp", address)
  194. if err != nil {
  195. return nil, fmt.Errorf("create server listener error, %v", err)
  196. }
  197. // Save listener so it can be closed in svr.Close()
  198. svr.tcpListener = ln
  199. svr.muxer = mux.NewMux(ln)
  200. svr.muxer.SetKeepAlive(time.Duration(cfg.Transport.TCPKeepAlive) * time.Second)
  201. go func() {
  202. _ = svr.muxer.Serve()
  203. }()
  204. ln = svr.muxer.DefaultListener()
  205. svr.muxListener = ln
  206. log.Infof("frps tcp listen on %s", address)
  207. // Listen for accepting connections from client using kcp protocol.
  208. if cfg.KCPBindPort > 0 {
  209. address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.KCPBindPort))
  210. svr.kcpListener, err = netpkg.ListenKcp(address)
  211. if err != nil {
  212. return nil, fmt.Errorf("listen on kcp udp address %s error: %v", address, err)
  213. }
  214. log.Infof("frps kcp listen on udp %s", address)
  215. }
  216. if cfg.QUICBindPort > 0 {
  217. address := net.JoinHostPort(cfg.BindAddr, strconv.Itoa(cfg.QUICBindPort))
  218. quicTLSCfg := tlsConfig.Clone()
  219. quicTLSCfg.NextProtos = []string{"frp"}
  220. svr.quicListener, err = quic.ListenAddr(address, quicTLSCfg, &quic.Config{
  221. MaxIdleTimeout: time.Duration(cfg.Transport.QUIC.MaxIdleTimeout) * time.Second,
  222. MaxIncomingStreams: int64(cfg.Transport.QUIC.MaxIncomingStreams),
  223. KeepAlivePeriod: time.Duration(cfg.Transport.QUIC.KeepalivePeriod) * time.Second,
  224. })
  225. if err != nil {
  226. return nil, fmt.Errorf("listen on quic udp address %s error: %v", address, err)
  227. }
  228. log.Infof("frps quic listen on %s", address)
  229. }
  230. if cfg.SSHTunnelGateway.BindPort > 0 {
  231. sshGateway, err := ssh.NewGateway(cfg.SSHTunnelGateway, cfg.ProxyBindAddr, svr.sshTunnelListener)
  232. if err != nil {
  233. return nil, fmt.Errorf("create ssh gateway error: %v", err)
  234. }
  235. svr.sshTunnelGateway = sshGateway
  236. log.Infof("frps sshTunnelGateway listen on port %d", cfg.SSHTunnelGateway.BindPort)
  237. }
  238. // Listen for accepting connections from client using websocket protocol.
  239. websocketPrefix := []byte("GET " + netpkg.FrpWebsocketPath)
  240. websocketLn := svr.muxer.Listen(0, uint32(len(websocketPrefix)), func(data []byte) bool {
  241. return bytes.Equal(data, websocketPrefix)
  242. })
  243. svr.websocketListener = netpkg.NewWebsocketListener(websocketLn)
  244. // Create http vhost muxer.
  245. if cfg.VhostHTTPPort > 0 {
  246. rp := vhost.NewHTTPReverseProxy(vhost.HTTPReverseProxyOptions{
  247. ResponseHeaderTimeoutS: cfg.VhostHTTPTimeout,
  248. }, svr.httpVhostRouter)
  249. svr.rc.HTTPReverseProxy = rp
  250. address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.VhostHTTPPort))
  251. server := &http.Server{
  252. Addr: address,
  253. Handler: rp,
  254. ReadHeaderTimeout: 60 * time.Second,
  255. }
  256. var l net.Listener
  257. if httpMuxOn {
  258. l = svr.muxer.ListenHTTP(1)
  259. } else {
  260. l, err = net.Listen("tcp", address)
  261. if err != nil {
  262. return nil, fmt.Errorf("create vhost http listener error, %v", err)
  263. }
  264. }
  265. go func() {
  266. _ = server.Serve(l)
  267. }()
  268. log.Infof("http service listen on %s", address)
  269. }
  270. // Create https vhost muxer.
  271. if cfg.VhostHTTPSPort > 0 {
  272. var l net.Listener
  273. if httpsMuxOn {
  274. l = svr.muxer.ListenHTTPS(1)
  275. } else {
  276. address := net.JoinHostPort(cfg.ProxyBindAddr, strconv.Itoa(cfg.VhostHTTPSPort))
  277. l, err = net.Listen("tcp", address)
  278. if err != nil {
  279. return nil, fmt.Errorf("create server listener error, %v", err)
  280. }
  281. log.Infof("https service listen on %s", address)
  282. }
  283. // Save listener so it can be closed in svr.Close()
  284. svr.vhostHTTPSListener = l
  285. svr.rc.VhostHTTPSMuxer, err = vhost.NewHTTPSMuxer(l, vhostReadWriteTimeout)
  286. if err != nil {
  287. return nil, fmt.Errorf("create vhost httpsMuxer error, %v", err)
  288. }
  289. }
  290. // frp tls listener
  291. svr.tlsListener = svr.muxer.Listen(2, 1, func(data []byte) bool {
  292. // tls first byte can be 0x16 only when vhost https port is not same with bind port
  293. return int(data[0]) == netpkg.FRPTLSHeadByte || int(data[0]) == 0x16
  294. })
  295. // Create nat hole controller.
  296. nc, err := nathole.NewController(time.Duration(cfg.NatHoleAnalysisDataReserveHours) * time.Hour)
  297. if err != nil {
  298. return nil, fmt.Errorf("create nat hole controller error, %v", err)
  299. }
  300. svr.rc.NatHoleController = nc
  301. return svr, nil
  302. }
  303. func (svr *Service) Run(ctx context.Context) {
  304. ctx, cancel := context.WithCancel(ctx)
  305. svr.ctx = ctx
  306. svr.cancel = cancel
  307. // run dashboard web server.
  308. if svr.webServer != nil {
  309. go func() {
  310. log.Infof("dashboard listen on %s", svr.webServer.Address())
  311. if err := svr.webServer.Run(); err != nil {
  312. log.Warnf("dashboard server exit with error: %v", err)
  313. }
  314. }()
  315. }
  316. go svr.HandleListener(svr.sshTunnelListener, true)
  317. if svr.kcpListener != nil {
  318. go svr.HandleListener(svr.kcpListener, false)
  319. }
  320. if svr.quicListener != nil {
  321. go svr.HandleQUICListener(svr.quicListener)
  322. }
  323. go svr.HandleListener(svr.websocketListener, false)
  324. go svr.HandleListener(svr.tlsListener, false)
  325. if svr.rc.NatHoleController != nil {
  326. go svr.rc.NatHoleController.CleanWorker(svr.ctx)
  327. }
  328. if svr.sshTunnelGateway != nil {
  329. go svr.sshTunnelGateway.Run()
  330. }
  331. svr.HandleListener(svr.muxListener, false)
  332. <-svr.ctx.Done()
  333. // service context may not be canceled by svr.Close(), we should call it here to release resources
  334. if svr.muxListener != nil {
  335. svr.Close()
  336. }
  337. }
  338. func (svr *Service) Close() error {
  339. if svr.kcpListener != nil {
  340. svr.kcpListener.Close()
  341. svr.kcpListener = nil
  342. }
  343. if svr.quicListener != nil {
  344. svr.quicListener.Close()
  345. svr.quicListener = nil
  346. }
  347. if svr.websocketListener != nil {
  348. svr.websocketListener.Close()
  349. svr.websocketListener = nil
  350. }
  351. if svr.tlsListener != nil {
  352. svr.tlsListener.Close()
  353. svr.tlsConfig = nil
  354. }
  355. if svr.muxListener != nil {
  356. svr.muxListener.Close()
  357. svr.muxListener = nil
  358. }
  359. if svr.vhostHTTPSListener != nil {
  360. svr.vhostHTTPSListener.Close()
  361. svr.vhostHTTPSListener = nil
  362. }
  363. if svr.tcpmuxHTTPConnectListener != nil {
  364. svr.tcpmuxHTTPConnectListener.Close()
  365. svr.tcpmuxHTTPConnectListener = nil
  366. }
  367. if svr.webServer != nil {
  368. svr.webServer.Close()
  369. svr.webServer = nil
  370. }
  371. if svr.tcpListener != nil {
  372. svr.tcpListener.Close()
  373. svr.tcpListener = nil
  374. }
  375. svr.ctlManager.Close()
  376. if svr.cancel != nil {
  377. svr.cancel()
  378. }
  379. return nil
  380. }
  381. func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, internal bool) {
  382. xl := xlog.FromContextSafe(ctx)
  383. var (
  384. rawMsg msg.Message
  385. err error
  386. )
  387. _ = conn.SetReadDeadline(time.Now().Add(connReadTimeout))
  388. if rawMsg, err = msg.ReadMsg(conn); err != nil {
  389. log.Tracef("Failed to read message: %v", err)
  390. conn.Close()
  391. return
  392. }
  393. _ = conn.SetReadDeadline(time.Time{})
  394. switch m := rawMsg.(type) {
  395. case *msg.Login:
  396. // server plugin hook
  397. content := &plugin.LoginContent{
  398. Login: *m,
  399. ClientAddress: conn.RemoteAddr().String(),
  400. }
  401. retContent, err := svr.pluginManager.Login(content)
  402. if err == nil {
  403. m = &retContent.Login
  404. err = svr.RegisterControl(conn, m, internal)
  405. }
  406. // If login failed, send error message there.
  407. // Otherwise send success message in control's work goroutine.
  408. if err != nil {
  409. xl.Warnf("register control error: %v", err)
  410. _ = msg.WriteMsg(conn, &msg.LoginResp{
  411. Version: version.Full(),
  412. Error: util.GenerateResponseErrorString("register control error", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
  413. })
  414. conn.Close()
  415. }
  416. case *msg.NewWorkConn:
  417. if err := svr.RegisterWorkConn(conn, m); err != nil {
  418. conn.Close()
  419. }
  420. case *msg.NewVisitorConn:
  421. if err = svr.RegisterVisitorConn(conn, m); err != nil {
  422. xl.Warnf("register visitor conn error: %v", err)
  423. _ = msg.WriteMsg(conn, &msg.NewVisitorConnResp{
  424. ProxyName: m.ProxyName,
  425. Error: util.GenerateResponseErrorString("register visitor conn error", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
  426. })
  427. conn.Close()
  428. } else {
  429. _ = msg.WriteMsg(conn, &msg.NewVisitorConnResp{
  430. ProxyName: m.ProxyName,
  431. Error: "",
  432. })
  433. }
  434. default:
  435. log.Warnf("Error message type for the new connection [%s]", conn.RemoteAddr().String())
  436. conn.Close()
  437. }
  438. }
  439. // HandleListener accepts connections from client and call handleConnection to handle them.
  440. // If internal is true, it means that this listener is used for internal communication like ssh tunnel gateway.
  441. // TODO(fatedier): Pass some parameters of listener/connection through context to avoid passing too many parameters.
  442. func (svr *Service) HandleListener(l net.Listener, internal bool) {
  443. // Listen for incoming connections from client.
  444. for {
  445. c, err := l.Accept()
  446. if err != nil {
  447. log.Warnf("Listener for incoming connections from client closed")
  448. return
  449. }
  450. // inject xlog object into net.Conn context
  451. xl := xlog.New()
  452. ctx := context.Background()
  453. c = netpkg.NewContextConn(xlog.NewContext(ctx, xl), c)
  454. if !internal {
  455. log.Tracef("start check TLS connection...")
  456. originConn := c
  457. forceTLS := svr.cfg.Transport.TLS.Force
  458. var isTLS, custom bool
  459. c, isTLS, custom, err = netpkg.CheckAndEnableTLSServerConnWithTimeout(c, svr.tlsConfig, forceTLS, connReadTimeout)
  460. if err != nil {
  461. log.Warnf("CheckAndEnableTLSServerConnWithTimeout error: %v", err)
  462. originConn.Close()
  463. continue
  464. }
  465. log.Tracef("check TLS connection success, isTLS: %v custom: %v internal: %v", isTLS, custom, internal)
  466. }
  467. // Start a new goroutine to handle connection.
  468. go func(ctx context.Context, frpConn net.Conn) {
  469. if lo.FromPtr(svr.cfg.Transport.TCPMux) && !internal {
  470. fmuxCfg := fmux.DefaultConfig()
  471. fmuxCfg.KeepAliveInterval = time.Duration(svr.cfg.Transport.TCPMuxKeepaliveInterval) * time.Second
  472. fmuxCfg.LogOutput = io.Discard
  473. fmuxCfg.MaxStreamWindowSize = 6 * 1024 * 1024
  474. session, err := fmux.Server(frpConn, fmuxCfg)
  475. if err != nil {
  476. log.Warnf("Failed to create mux connection: %v", err)
  477. frpConn.Close()
  478. return
  479. }
  480. for {
  481. stream, err := session.AcceptStream()
  482. if err != nil {
  483. log.Debugf("Accept new mux stream error: %v", err)
  484. session.Close()
  485. return
  486. }
  487. go svr.handleConnection(ctx, stream, internal)
  488. }
  489. } else {
  490. svr.handleConnection(ctx, frpConn, internal)
  491. }
  492. }(ctx, c)
  493. }
  494. }
  495. func (svr *Service) HandleQUICListener(l *quic.Listener) {
  496. // Listen for incoming connections from client.
  497. for {
  498. c, err := l.Accept(context.Background())
  499. if err != nil {
  500. log.Warnf("QUICListener for incoming connections from client closed")
  501. return
  502. }
  503. // Start a new goroutine to handle connection.
  504. go func(ctx context.Context, frpConn quic.Connection) {
  505. for {
  506. stream, err := frpConn.AcceptStream(context.Background())
  507. if err != nil {
  508. log.Debugf("Accept new quic mux stream error: %v", err)
  509. _ = frpConn.CloseWithError(0, "")
  510. return
  511. }
  512. go svr.handleConnection(ctx, netpkg.QuicStreamToNetConn(stream, frpConn), false)
  513. }
  514. }(context.Background(), c)
  515. }
  516. }
  517. func (svr *Service) RegisterControl(ctlConn net.Conn, loginMsg *msg.Login, internal bool) error {
  518. // If client's RunID is empty, it's a new client, we just create a new controller.
  519. // Otherwise, we check if there is one controller has the same run id. If so, we release previous controller and start new one.
  520. var err error
  521. if loginMsg.RunID == "" {
  522. loginMsg.RunID, err = util.RandID()
  523. if err != nil {
  524. return err
  525. }
  526. }
  527. ctx := netpkg.NewContextFromConn(ctlConn)
  528. xl := xlog.FromContextSafe(ctx)
  529. xl.AppendPrefix(loginMsg.RunID)
  530. ctx = xlog.NewContext(ctx, xl)
  531. xl.Infof("client login info: ip [%s] version [%s] hostname [%s] os [%s] arch [%s]",
  532. ctlConn.RemoteAddr().String(), loginMsg.Version, loginMsg.Hostname, loginMsg.Os, loginMsg.Arch)
  533. // Check auth.
  534. authVerifier := svr.authVerifier
  535. if internal && loginMsg.ClientSpec.AlwaysAuthPass {
  536. authVerifier = auth.AlwaysPassVerifier
  537. }
  538. if err := authVerifier.VerifyLogin(loginMsg); err != nil {
  539. return err
  540. }
  541. // TODO(fatedier): use SessionContext
  542. ctl, err := NewControl(ctx, svr.rc, svr.pxyManager, svr.pluginManager, authVerifier, ctlConn, !internal, loginMsg, svr.cfg)
  543. if err != nil {
  544. xl.Warnf("create new controller error: %v", err)
  545. // don't return detailed errors to client
  546. return fmt.Errorf("unexpected error when creating new controller")
  547. }
  548. if oldCtl := svr.ctlManager.Add(loginMsg.RunID, ctl); oldCtl != nil {
  549. oldCtl.WaitClosed()
  550. }
  551. ctl.Start()
  552. // for statistics
  553. metrics.Server.NewClient()
  554. go func() {
  555. // block until control closed
  556. ctl.WaitClosed()
  557. svr.ctlManager.Del(loginMsg.RunID, ctl)
  558. }()
  559. return nil
  560. }
  561. // RegisterWorkConn register a new work connection to control and proxies need it.
  562. func (svr *Service) RegisterWorkConn(workConn net.Conn, newMsg *msg.NewWorkConn) error {
  563. xl := netpkg.NewLogFromConn(workConn)
  564. ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
  565. if !exist {
  566. xl.Warnf("No client control found for run id [%s]", newMsg.RunID)
  567. return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
  568. }
  569. // server plugin hook
  570. content := &plugin.NewWorkConnContent{
  571. User: plugin.UserInfo{
  572. User: ctl.loginMsg.User,
  573. Metas: ctl.loginMsg.Metas,
  574. RunID: ctl.loginMsg.RunID,
  575. },
  576. NewWorkConn: *newMsg,
  577. }
  578. retContent, err := svr.pluginManager.NewWorkConn(content)
  579. if err == nil {
  580. newMsg = &retContent.NewWorkConn
  581. // Check auth.
  582. err = ctl.authVerifier.VerifyNewWorkConn(newMsg)
  583. }
  584. if err != nil {
  585. xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
  586. _ = msg.WriteMsg(workConn, &msg.StartWorkConn{
  587. Error: util.GenerateResponseErrorString("invalid NewWorkConn", err, lo.FromPtr(svr.cfg.DetailedErrorsToClient)),
  588. })
  589. return fmt.Errorf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
  590. }
  591. return ctl.RegisterWorkConn(workConn)
  592. }
  593. func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVisitorConn) error {
  594. visitorUser := ""
  595. // TODO(deprecation): Compatible with old versions, can be without runID, user is empty. In later versions, it will be mandatory to include runID.
  596. // If runID is required, it is not compatible with versions prior to v0.50.0.
  597. if newMsg.RunID != "" {
  598. ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
  599. if !exist {
  600. return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
  601. }
  602. visitorUser = ctl.loginMsg.User
  603. }
  604. return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
  605. newMsg.UseEncryption, newMsg.UseCompression, visitorUser)
  606. }