visitor.go 7.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322
  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. "bytes"
  17. "fmt"
  18. "io"
  19. "net"
  20. "strconv"
  21. "strings"
  22. "sync"
  23. "time"
  24. "golang.org/x/net/ipv4"
  25. "github.com/fatedier/frp/models/config"
  26. "github.com/fatedier/frp/models/msg"
  27. frpIo "github.com/fatedier/frp/utils/io"
  28. "github.com/fatedier/frp/utils/log"
  29. frpNet "github.com/fatedier/frp/utils/net"
  30. "github.com/fatedier/frp/utils/pool"
  31. "github.com/fatedier/frp/utils/util"
  32. )
  33. // Visitor is used for forward traffics from local port tot remote service.
  34. type Visitor interface {
  35. Run() error
  36. Close()
  37. log.Logger
  38. }
  39. func NewVisitor(ctl *Control, pxyConf config.ProxyConf) (visitor Visitor) {
  40. baseVisitor := BaseVisitor{
  41. ctl: ctl,
  42. Logger: log.NewPrefixLogger(pxyConf.GetName()),
  43. }
  44. switch cfg := pxyConf.(type) {
  45. case *config.StcpProxyConf:
  46. visitor = &StcpVisitor{
  47. BaseVisitor: baseVisitor,
  48. cfg: cfg,
  49. }
  50. case *config.XtcpProxyConf:
  51. visitor = &XtcpVisitor{
  52. BaseVisitor: baseVisitor,
  53. cfg: cfg,
  54. }
  55. }
  56. return
  57. }
  58. type BaseVisitor struct {
  59. ctl *Control
  60. l frpNet.Listener
  61. closed bool
  62. mu sync.RWMutex
  63. log.Logger
  64. }
  65. type StcpVisitor struct {
  66. BaseVisitor
  67. cfg *config.StcpProxyConf
  68. }
  69. func (sv *StcpVisitor) Run() (err error) {
  70. sv.l, err = frpNet.ListenTcp(sv.cfg.BindAddr, sv.cfg.BindPort)
  71. if err != nil {
  72. return
  73. }
  74. go sv.worker()
  75. return
  76. }
  77. func (sv *StcpVisitor) Close() {
  78. sv.l.Close()
  79. }
  80. func (sv *StcpVisitor) worker() {
  81. for {
  82. conn, err := sv.l.Accept()
  83. if err != nil {
  84. sv.Warn("stcp local listener closed")
  85. return
  86. }
  87. go sv.handleConn(conn)
  88. }
  89. }
  90. func (sv *StcpVisitor) handleConn(userConn frpNet.Conn) {
  91. defer userConn.Close()
  92. sv.Debug("get a new stcp user connection")
  93. visitorConn, err := sv.ctl.connectServer()
  94. if err != nil {
  95. return
  96. }
  97. defer visitorConn.Close()
  98. now := time.Now().Unix()
  99. newVisitorConnMsg := &msg.NewVisitorConn{
  100. ProxyName: sv.cfg.ServerName,
  101. SignKey: util.GetAuthKey(sv.cfg.Sk, now),
  102. Timestamp: now,
  103. UseEncryption: sv.cfg.UseEncryption,
  104. UseCompression: sv.cfg.UseCompression,
  105. }
  106. err = msg.WriteMsg(visitorConn, newVisitorConnMsg)
  107. if err != nil {
  108. sv.Warn("send newVisitorConnMsg to server error: %v", err)
  109. return
  110. }
  111. var newVisitorConnRespMsg msg.NewVisitorConnResp
  112. visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
  113. err = msg.ReadMsgInto(visitorConn, &newVisitorConnRespMsg)
  114. if err != nil {
  115. sv.Warn("get newVisitorConnRespMsg error: %v", err)
  116. return
  117. }
  118. visitorConn.SetReadDeadline(time.Time{})
  119. if newVisitorConnRespMsg.Error != "" {
  120. sv.Warn("start new visitor connection error: %s", newVisitorConnRespMsg.Error)
  121. return
  122. }
  123. var remote io.ReadWriteCloser
  124. remote = visitorConn
  125. if sv.cfg.UseEncryption {
  126. remote, err = frpIo.WithEncryption(remote, []byte(sv.cfg.Sk))
  127. if err != nil {
  128. sv.Error("create encryption stream error: %v", err)
  129. return
  130. }
  131. }
  132. if sv.cfg.UseCompression {
  133. remote = frpIo.WithCompression(remote)
  134. }
  135. frpIo.Join(userConn, remote)
  136. }
  137. type XtcpVisitor struct {
  138. BaseVisitor
  139. cfg *config.XtcpProxyConf
  140. }
  141. func (sv *XtcpVisitor) Run() (err error) {
  142. sv.l, err = frpNet.ListenTcp(sv.cfg.BindAddr, sv.cfg.BindPort)
  143. if err != nil {
  144. return
  145. }
  146. go sv.worker()
  147. return
  148. }
  149. func (sv *XtcpVisitor) Close() {
  150. sv.l.Close()
  151. }
  152. func (sv *XtcpVisitor) worker() {
  153. for {
  154. conn, err := sv.l.Accept()
  155. if err != nil {
  156. sv.Warn("stcp local listener closed")
  157. return
  158. }
  159. go sv.handleConn(conn)
  160. }
  161. }
  162. func (sv *XtcpVisitor) handleConn(userConn frpNet.Conn) {
  163. defer userConn.Close()
  164. sv.Debug("get a new xtcp user connection")
  165. if config.ClientCommonCfg.ServerUdpPort == 0 {
  166. sv.Error("xtcp is not supported by server")
  167. return
  168. }
  169. raddr, err := net.ResolveUDPAddr("udp",
  170. fmt.Sprintf("%s:%d", config.ClientCommonCfg.ServerAddr, config.ClientCommonCfg.ServerUdpPort))
  171. visitorConn, err := net.DialUDP("udp", nil, raddr)
  172. defer visitorConn.Close()
  173. now := time.Now().Unix()
  174. natHoleVisitorMsg := &msg.NatHoleVisitor{
  175. ProxyName: sv.cfg.ServerName,
  176. SignKey: util.GetAuthKey(sv.cfg.Sk, now),
  177. Timestamp: now,
  178. }
  179. err = msg.WriteMsg(visitorConn, natHoleVisitorMsg)
  180. if err != nil {
  181. sv.Warn("send natHoleVisitorMsg to server error: %v", err)
  182. return
  183. }
  184. // Wait for client address at most 10 seconds.
  185. var natHoleRespMsg msg.NatHoleResp
  186. visitorConn.SetReadDeadline(time.Now().Add(10 * time.Second))
  187. buf := pool.GetBuf(1024)
  188. n, err := visitorConn.Read(buf)
  189. if err != nil {
  190. sv.Warn("get natHoleRespMsg error: %v", err)
  191. return
  192. }
  193. err = msg.ReadMsgInto(bytes.NewReader(buf[:n]), &natHoleRespMsg)
  194. if err != nil {
  195. sv.Warn("get natHoleRespMsg error: %v", err)
  196. return
  197. }
  198. visitorConn.SetReadDeadline(time.Time{})
  199. pool.PutBuf(buf)
  200. sv.Trace("get natHoleRespMsg, sid [%s], client address [%s]", natHoleRespMsg.Sid, natHoleRespMsg.ClientAddr)
  201. // Close visitorConn, so we can use it's local address.
  202. visitorConn.Close()
  203. // Send detect message.
  204. array := strings.Split(natHoleRespMsg.ClientAddr, ":")
  205. if len(array) <= 1 {
  206. sv.Error("get natHoleResp client address error: %s", natHoleRespMsg.ClientAddr)
  207. return
  208. }
  209. laddr, _ := net.ResolveUDPAddr("udp", visitorConn.LocalAddr().String())
  210. /*
  211. for i := 1000; i < 65000; i++ {
  212. sv.sendDetectMsg(array[0], int64(i), laddr, "a")
  213. }
  214. */
  215. port, err := strconv.ParseInt(array[1], 10, 64)
  216. if err != nil {
  217. sv.Error("get natHoleResp client address error: %s", natHoleRespMsg.ClientAddr)
  218. return
  219. }
  220. sv.sendDetectMsg(array[0], int(port), laddr, []byte(natHoleRespMsg.Sid))
  221. sv.Trace("send all detect msg done")
  222. // Listen for visitorConn's address and wait for client connection.
  223. lConn, err := net.ListenUDP("udp", laddr)
  224. if err != nil {
  225. sv.Error("listen on visitorConn's local adress error: %v", err)
  226. return
  227. }
  228. lConn.SetReadDeadline(time.Now().Add(5 * time.Second))
  229. sidBuf := pool.GetBuf(1024)
  230. n, _, err = lConn.ReadFromUDP(sidBuf)
  231. if err != nil {
  232. sv.Warn("get sid from client error: %v", err)
  233. return
  234. }
  235. lConn.SetReadDeadline(time.Time{})
  236. if string(sidBuf[:n]) != natHoleRespMsg.Sid {
  237. sv.Warn("incorrect sid from client")
  238. return
  239. }
  240. sv.Info("nat hole connection make success, sid [%s]", string(sidBuf[:n]))
  241. pool.PutBuf(sidBuf)
  242. var remote io.ReadWriteCloser
  243. remote, err = frpNet.NewKcpConnFromUdp(lConn, false, natHoleRespMsg.ClientAddr)
  244. if err != nil {
  245. sv.Error("create kcp connection from udp connection error: %v", err)
  246. return
  247. }
  248. if sv.cfg.UseEncryption {
  249. remote, err = frpIo.WithEncryption(remote, []byte(sv.cfg.Sk))
  250. if err != nil {
  251. sv.Error("create encryption stream error: %v", err)
  252. return
  253. }
  254. }
  255. if sv.cfg.UseCompression {
  256. remote = frpIo.WithCompression(remote)
  257. }
  258. frpIo.Join(userConn, remote)
  259. sv.Debug("join connections closed")
  260. }
  261. func (sv *XtcpVisitor) sendDetectMsg(addr string, port int, laddr *net.UDPAddr, content []byte) (err error) {
  262. daddr, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", addr, port))
  263. if err != nil {
  264. return err
  265. }
  266. tConn, err := net.DialUDP("udp", laddr, daddr)
  267. if err != nil {
  268. return err
  269. }
  270. uConn := ipv4.NewConn(tConn)
  271. uConn.SetTTL(3)
  272. tConn.Write(content)
  273. tConn.Close()
  274. return nil
  275. }