service.go 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323
  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 client
  15. import (
  16. "context"
  17. "crypto/tls"
  18. "errors"
  19. "fmt"
  20. "io/ioutil"
  21. "net"
  22. "runtime"
  23. "strconv"
  24. "sync"
  25. "sync/atomic"
  26. "time"
  27. "github.com/fatedier/frp/assets"
  28. "github.com/fatedier/frp/pkg/auth"
  29. "github.com/fatedier/frp/pkg/config"
  30. "github.com/fatedier/frp/pkg/msg"
  31. "github.com/fatedier/frp/pkg/transport"
  32. "github.com/fatedier/frp/pkg/util/log"
  33. frpNet "github.com/fatedier/frp/pkg/util/net"
  34. "github.com/fatedier/frp/pkg/util/version"
  35. "github.com/fatedier/frp/pkg/util/xlog"
  36. fmux "github.com/hashicorp/yamux"
  37. )
  38. // Service is a client service.
  39. type Service struct {
  40. // uniq id got from frps, attach it in loginMsg
  41. runID string
  42. // manager control connection with server
  43. ctl *Control
  44. ctlMu sync.RWMutex
  45. // Sets authentication based on selected method
  46. authSetter auth.Setter
  47. cfg config.ClientCommonConf
  48. pxyCfgs map[string]config.ProxyConf
  49. visitorCfgs map[string]config.VisitorConf
  50. cfgMu sync.RWMutex
  51. // The configuration file used to initialize this client, or an empty
  52. // string if no configuration file was used.
  53. cfgFile string
  54. // This is configured by the login response from frps
  55. serverUDPPort int
  56. exit uint32 // 0 means not exit
  57. // service context
  58. ctx context.Context
  59. // call cancel to stop service
  60. cancel context.CancelFunc
  61. }
  62. func NewService(cfg config.ClientCommonConf, pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf, cfgFile string) (svr *Service, err error) {
  63. ctx, cancel := context.WithCancel(context.Background())
  64. svr = &Service{
  65. authSetter: auth.NewAuthSetter(cfg.ClientConfig),
  66. cfg: cfg,
  67. cfgFile: cfgFile,
  68. pxyCfgs: pxyCfgs,
  69. visitorCfgs: visitorCfgs,
  70. exit: 0,
  71. ctx: xlog.NewContext(ctx, xlog.New()),
  72. cancel: cancel,
  73. }
  74. return
  75. }
  76. func (svr *Service) GetController() *Control {
  77. svr.ctlMu.RLock()
  78. defer svr.ctlMu.RUnlock()
  79. return svr.ctl
  80. }
  81. func (svr *Service) Run() error {
  82. xl := xlog.FromContextSafe(svr.ctx)
  83. // login to frps
  84. for {
  85. conn, session, err := svr.login()
  86. if err != nil {
  87. xl.Warn("login to server failed: %v", err)
  88. // if login_fail_exit is true, just exit this program
  89. // otherwise sleep a while and try again to connect to server
  90. if svr.cfg.LoginFailExit {
  91. return err
  92. }
  93. time.Sleep(10 * time.Second)
  94. } else {
  95. // login success
  96. ctl := NewControl(svr.ctx, svr.runID, conn, session, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.serverUDPPort, svr.authSetter)
  97. ctl.Run()
  98. svr.ctlMu.Lock()
  99. svr.ctl = ctl
  100. svr.ctlMu.Unlock()
  101. break
  102. }
  103. }
  104. go svr.keepControllerWorking()
  105. if svr.cfg.AdminPort != 0 {
  106. // Init admin server assets
  107. err := assets.Load(svr.cfg.AssetsDir)
  108. if err != nil {
  109. return fmt.Errorf("Load assets error: %v", err)
  110. }
  111. address := net.JoinHostPort(svr.cfg.AdminAddr, strconv.Itoa(svr.cfg.AdminPort))
  112. err = svr.RunAdminServer(address)
  113. if err != nil {
  114. log.Warn("run admin server error: %v", err)
  115. }
  116. log.Info("admin server listen on %s:%d", svr.cfg.AdminAddr, svr.cfg.AdminPort)
  117. }
  118. <-svr.ctx.Done()
  119. return nil
  120. }
  121. func (svr *Service) keepControllerWorking() {
  122. xl := xlog.FromContextSafe(svr.ctx)
  123. maxDelayTime := 20 * time.Second
  124. delayTime := time.Second
  125. // if frpc reconnect frps, we need to limit retry times in 1min
  126. // current retry logic is sleep 0s, 0s, 0s, 1s, 2s, 4s, 8s, ...
  127. // when exceed 1min, we will reset delay and counts
  128. cutoffTime := time.Now().Add(time.Minute)
  129. reconnectDelay := time.Second
  130. reconnectCounts := 1
  131. for {
  132. <-svr.ctl.ClosedDoneCh()
  133. if atomic.LoadUint32(&svr.exit) != 0 {
  134. return
  135. }
  136. // the first three retry with no delay
  137. if reconnectCounts > 3 {
  138. time.Sleep(reconnectDelay)
  139. reconnectDelay *= 2
  140. }
  141. reconnectCounts++
  142. now := time.Now()
  143. if now.After(cutoffTime) {
  144. // reset
  145. cutoffTime = now.Add(time.Minute)
  146. reconnectDelay = time.Second
  147. reconnectCounts = 1
  148. }
  149. for {
  150. xl.Info("try to reconnect to server...")
  151. conn, session, err := svr.login()
  152. if err != nil {
  153. xl.Warn("reconnect to server error: %v", err)
  154. time.Sleep(delayTime)
  155. opErr := &net.OpError{}
  156. // quick retry for dial error
  157. if errors.As(err, &opErr) && opErr.Op == "dial" {
  158. delayTime = 2 * time.Second
  159. } else {
  160. delayTime = delayTime * 2
  161. if delayTime > maxDelayTime {
  162. delayTime = maxDelayTime
  163. }
  164. }
  165. continue
  166. }
  167. // reconnect success, init delayTime
  168. delayTime = time.Second
  169. ctl := NewControl(svr.ctx, svr.runID, conn, session, svr.cfg, svr.pxyCfgs, svr.visitorCfgs, svr.serverUDPPort, svr.authSetter)
  170. ctl.Run()
  171. svr.ctlMu.Lock()
  172. if svr.ctl != nil {
  173. svr.ctl.Close()
  174. }
  175. svr.ctl = ctl
  176. svr.ctlMu.Unlock()
  177. break
  178. }
  179. }
  180. }
  181. // login creates a connection to frps and registers it self as a client
  182. // conn: control connection
  183. // session: if it's not nil, using tcp mux
  184. func (svr *Service) login() (conn net.Conn, session *fmux.Session, err error) {
  185. xl := xlog.FromContextSafe(svr.ctx)
  186. var tlsConfig *tls.Config
  187. if svr.cfg.TLSEnable {
  188. sn := svr.cfg.TLSServerName
  189. if sn == "" {
  190. sn = svr.cfg.ServerAddr
  191. }
  192. tlsConfig, err = transport.NewClientTLSConfig(
  193. svr.cfg.TLSCertFile,
  194. svr.cfg.TLSKeyFile,
  195. svr.cfg.TLSTrustedCaFile,
  196. sn)
  197. if err != nil {
  198. xl.Warn("fail to build tls configuration when service login, err: %v", err)
  199. return
  200. }
  201. }
  202. address := net.JoinHostPort(svr.cfg.ServerAddr, strconv.Itoa(svr.cfg.ServerPort))
  203. conn, err = frpNet.ConnectServerByProxyWithTLS(svr.cfg.HTTPProxy, svr.cfg.Protocol, address, tlsConfig, svr.cfg.DisableCustomTLSFirstByte)
  204. if err != nil {
  205. return
  206. }
  207. defer func() {
  208. if err != nil {
  209. conn.Close()
  210. if session != nil {
  211. session.Close()
  212. }
  213. }
  214. }()
  215. if svr.cfg.TCPMux {
  216. fmuxCfg := fmux.DefaultConfig()
  217. fmuxCfg.KeepAliveInterval = 20 * time.Second
  218. fmuxCfg.LogOutput = ioutil.Discard
  219. session, err = fmux.Client(conn, fmuxCfg)
  220. if err != nil {
  221. return
  222. }
  223. stream, errRet := session.OpenStream()
  224. if errRet != nil {
  225. session.Close()
  226. err = errRet
  227. return
  228. }
  229. conn = stream
  230. }
  231. loginMsg := &msg.Login{
  232. Arch: runtime.GOARCH,
  233. Os: runtime.GOOS,
  234. PoolCount: svr.cfg.PoolCount,
  235. User: svr.cfg.User,
  236. Version: version.Full(),
  237. Timestamp: time.Now().Unix(),
  238. RunID: svr.runID,
  239. Metas: svr.cfg.Metas,
  240. }
  241. // Add auth
  242. if err = svr.authSetter.SetLogin(loginMsg); err != nil {
  243. return
  244. }
  245. if err = msg.WriteMsg(conn, loginMsg); err != nil {
  246. return
  247. }
  248. var loginRespMsg msg.LoginResp
  249. conn.SetReadDeadline(time.Now().Add(10 * time.Second))
  250. if err = msg.ReadMsgInto(conn, &loginRespMsg); err != nil {
  251. return
  252. }
  253. conn.SetReadDeadline(time.Time{})
  254. if loginRespMsg.Error != "" {
  255. err = fmt.Errorf("%s", loginRespMsg.Error)
  256. xl.Error("%s", loginRespMsg.Error)
  257. return
  258. }
  259. svr.runID = loginRespMsg.RunID
  260. xl.ResetPrefixes()
  261. xl.AppendPrefix(svr.runID)
  262. svr.serverUDPPort = loginRespMsg.ServerUDPPort
  263. xl.Info("login to server success, get run id [%s], server udp port [%d]", loginRespMsg.RunID, loginRespMsg.ServerUDPPort)
  264. return
  265. }
  266. func (svr *Service) ReloadConf(pxyCfgs map[string]config.ProxyConf, visitorCfgs map[string]config.VisitorConf) error {
  267. svr.cfgMu.Lock()
  268. svr.pxyCfgs = pxyCfgs
  269. svr.visitorCfgs = visitorCfgs
  270. svr.cfgMu.Unlock()
  271. return svr.ctl.ReloadConf(pxyCfgs, visitorCfgs)
  272. }
  273. func (svr *Service) Close() {
  274. atomic.StoreUint32(&svr.exit, 1)
  275. if svr.ctl != nil {
  276. svr.ctl.Close()
  277. }
  278. svr.cancel()
  279. }