conn.go 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358
  1. package tcpprox
  2. import (
  3. "errors"
  4. "io"
  5. "log"
  6. "net"
  7. "strconv"
  8. "sync"
  9. "sync/atomic"
  10. "time"
  11. )
  12. func isValidIP(ip string) bool {
  13. parsedIP := net.ParseIP(ip)
  14. return parsedIP != nil
  15. }
  16. func isValidPort(port string) bool {
  17. portInt, err := strconv.Atoi(port)
  18. if err != nil {
  19. return false
  20. }
  21. if portInt < 1 || portInt > 65535 {
  22. return false
  23. }
  24. return true
  25. }
  26. func isReachable(target string) bool {
  27. timeout := time.Duration(2 * time.Second) // Set the timeout value as per your requirement
  28. conn, err := net.DialTimeout("tcp", target, timeout)
  29. if err != nil {
  30. return false
  31. }
  32. defer conn.Close()
  33. return true
  34. }
  35. func connCopy(conn1 net.Conn, conn2 net.Conn, wg *sync.WaitGroup, accumulator *atomic.Int64) {
  36. n, err := io.Copy(conn1, conn2)
  37. if err != nil {
  38. //Add to accumulator
  39. accumulator.Add(n)
  40. }
  41. conn1.Close()
  42. log.Println("[←]", "close the connect at local:["+conn1.LocalAddr().String()+"] and remote:["+conn1.RemoteAddr().String()+"]")
  43. //conn2.Close()
  44. //log.Println("[←]", "close the connect at local:["+conn2.LocalAddr().String()+"] and remote:["+conn2.RemoteAddr().String()+"]")
  45. wg.Done()
  46. }
  47. func forward(conn1 net.Conn, conn2 net.Conn, aTob *atomic.Int64, bToa *atomic.Int64) {
  48. log.Printf("[+] start transmit. [%s],[%s] <-> [%s],[%s] \n", conn1.LocalAddr().String(), conn1.RemoteAddr().String(), conn2.LocalAddr().String(), conn2.RemoteAddr().String())
  49. var wg sync.WaitGroup
  50. // wait tow goroutines
  51. wg.Add(2)
  52. go connCopy(conn1, conn2, &wg, aTob)
  53. go connCopy(conn2, conn1, &wg, bToa)
  54. //blocking when the wg is locked
  55. wg.Wait()
  56. }
  57. func (c *ProxyRelayConfig) accept(listener net.Listener) (net.Conn, error) {
  58. conn, err := listener.Accept()
  59. if err != nil {
  60. return nil, err
  61. }
  62. //Check if connection in blacklist or whitelist
  63. if addr, ok := conn.RemoteAddr().(*net.TCPAddr); ok {
  64. if !c.parent.Options.AccessControlHandler(conn) {
  65. time.Sleep(300 * time.Millisecond)
  66. conn.Close()
  67. log.Println("[x]", "Connection from "+addr.IP.String()+" rejected by access control policy")
  68. return nil, errors.New("Connection from " + addr.IP.String() + " rejected by access control policy")
  69. }
  70. }
  71. log.Println("[√]", "accept a new client. remote address:["+conn.RemoteAddr().String()+"], local address:["+conn.LocalAddr().String()+"]")
  72. return conn, err
  73. }
  74. func startListener(address string) (net.Listener, error) {
  75. log.Println("[+]", "try to start server on:["+address+"]")
  76. server, err := net.Listen("tcp", address)
  77. if err != nil {
  78. return nil, errors.New("listen address [" + address + "] faild")
  79. }
  80. log.Println("[√]", "start listen at address:["+address+"]")
  81. return server, nil
  82. }
  83. /*
  84. Config Functions
  85. */
  86. // Config validator
  87. func (c *ProxyRelayConfig) ValidateConfigs() error {
  88. if c.Mode == ProxyMode_Transport {
  89. //Port2Host: PortA int, PortB string
  90. if !isValidPort(c.PortA) {
  91. return errors.New("first address must be a valid port number")
  92. }
  93. if !isReachable(c.PortB) {
  94. return errors.New("second address is unreachable")
  95. }
  96. return nil
  97. } else if c.Mode == ProxyMode_Listen {
  98. //Port2Port: Both port are port number
  99. if !isValidPort(c.PortA) {
  100. return errors.New("first address is not a valid port number")
  101. }
  102. if !isValidPort(c.PortB) {
  103. return errors.New("second address is not a valid port number")
  104. }
  105. return nil
  106. } else if c.Mode == ProxyMode_Starter {
  107. //Host2Host: Both have to be hosts
  108. if !isReachable(c.PortA) {
  109. return errors.New("first address is unreachable")
  110. }
  111. if !isReachable(c.PortB) {
  112. return errors.New("second address is unreachable")
  113. }
  114. return nil
  115. } else if c.Mode == ProxyMode_UDP {
  116. //UDP
  117. if !isValidPort(c.PortA) {
  118. return errors.New("first address must be a valid port number")
  119. }
  120. return nil
  121. }
  122. return errors.New("invalid mode given")
  123. }
  124. // Start a proxy if stopped
  125. func (c *ProxyRelayConfig) Start() error {
  126. if c.Running {
  127. return errors.New("proxy already running")
  128. }
  129. // Create a stopChan to control the loop
  130. stopChan := make(chan bool)
  131. c.stopChan = stopChan
  132. //Validate configs
  133. err := c.ValidateConfigs()
  134. if err != nil {
  135. return err
  136. }
  137. //Start the proxy service
  138. go func() {
  139. c.Running = true
  140. if c.Mode == ProxyMode_Transport {
  141. err = c.Port2host(c.PortA, c.PortB, stopChan)
  142. } else if c.Mode == ProxyMode_Listen {
  143. err = c.Port2port(c.PortA, c.PortB, stopChan)
  144. } else if c.Mode == ProxyMode_Starter {
  145. err = c.Host2host(c.PortA, c.PortB, stopChan)
  146. } else if c.Mode == ProxyMode_UDP {
  147. err = c.ForwardUDP(c.PortA, c.PortB, stopChan)
  148. }
  149. if err != nil {
  150. c.Running = false
  151. log.Println("Error starting proxy service " + c.Name + "(" + c.UUID + "): " + err.Error())
  152. }
  153. }()
  154. //Successfully spawned off the proxy routine
  155. return nil
  156. }
  157. // Stop a running proxy if running
  158. func (c *ProxyRelayConfig) IsRunning() bool {
  159. return c.Running || c.stopChan != nil
  160. }
  161. // Stop a running proxy if running
  162. func (c *ProxyRelayConfig) Stop() {
  163. log.Println("[PROXY] Stopping Stream Proxy " + c.Name)
  164. if c.Running || c.stopChan != nil {
  165. c.stopChan <- true
  166. time.Sleep(300 * time.Millisecond)
  167. c.stopChan = nil
  168. c.Running = false
  169. }
  170. log.Println("[PROXY] Stopped Stream Proxy " + c.Name)
  171. }
  172. /*
  173. Forwarder Functions
  174. */
  175. /*
  176. portA -> server
  177. portB -> server
  178. */
  179. func (c *ProxyRelayConfig) Port2port(port1 string, port2 string, stopChan chan bool) error {
  180. //Trim the Prefix of : if exists
  181. listen1, err := startListener("0.0.0.0:" + port1)
  182. if err != nil {
  183. return err
  184. }
  185. listen2, err := startListener("0.0.0.0:" + port2)
  186. if err != nil {
  187. return err
  188. }
  189. log.Println("[√]", "listen port:", port1, "and", port2, "success. waiting for client...")
  190. c.Running = true
  191. go func() {
  192. <-stopChan
  193. log.Println("[x]", "Received stop signal. Exiting Port to Port forwarder")
  194. c.Running = false
  195. listen1.Close()
  196. listen2.Close()
  197. }()
  198. for {
  199. conn1, err := c.accept(listen1)
  200. if err != nil {
  201. if !c.Running {
  202. return nil
  203. }
  204. continue
  205. }
  206. conn2, err := c.accept(listen2)
  207. if err != nil {
  208. if !c.Running {
  209. return nil
  210. }
  211. continue
  212. }
  213. if conn1 == nil || conn2 == nil {
  214. log.Println("[x]", "accept client faild. retry in ", c.Timeout, " seconds. ")
  215. time.Sleep(time.Duration(c.Timeout) * time.Second)
  216. continue
  217. }
  218. go forward(conn1, conn2, &c.aTobAccumulatedByteTransfer, &c.bToaAccumulatedByteTransfer)
  219. }
  220. }
  221. /*
  222. portA -> server
  223. server -> portB
  224. */
  225. func (c *ProxyRelayConfig) Port2host(allowPort string, targetAddress string, stopChan chan bool) error {
  226. server, err := startListener("0.0.0.0:" + allowPort)
  227. if err != nil {
  228. return err
  229. }
  230. //Start stop handler
  231. go func() {
  232. <-stopChan
  233. log.Println("[x]", "Received stop signal. Exiting Port to Host forwarder")
  234. c.Running = false
  235. server.Close()
  236. }()
  237. //Start blocking loop for accepting connections
  238. for {
  239. conn, err := c.accept(server)
  240. if conn == nil || err != nil {
  241. if !c.Running {
  242. //Terminate by stop chan. Exit listener loop
  243. return nil
  244. }
  245. //Connection error. Retry
  246. continue
  247. }
  248. go func(targetAddress string) {
  249. log.Println("[+]", "start connect host:["+targetAddress+"]")
  250. target, err := net.Dial("tcp", targetAddress)
  251. if err != nil {
  252. // temporarily unavailable, don't use fatal.
  253. log.Println("[x]", "connect target address ["+targetAddress+"] faild. retry in ", c.Timeout, "seconds. ")
  254. conn.Close()
  255. log.Println("[←]", "close the connect at local:["+conn.LocalAddr().String()+"] and remote:["+conn.RemoteAddr().String()+"]")
  256. time.Sleep(time.Duration(c.Timeout) * time.Second)
  257. return
  258. }
  259. log.Println("[→]", "connect target address ["+targetAddress+"] success.")
  260. forward(target, conn, &c.aTobAccumulatedByteTransfer, &c.bToaAccumulatedByteTransfer)
  261. }(targetAddress)
  262. }
  263. }
  264. /*
  265. server -> portA
  266. server -> portB
  267. */
  268. func (c *ProxyRelayConfig) Host2host(address1, address2 string, stopChan chan bool) error {
  269. c.Running = true
  270. go func() {
  271. <-stopChan
  272. log.Println("[x]", "Received stop signal. Exiting Host to Host forwarder")
  273. c.Running = false
  274. }()
  275. for c.Running {
  276. log.Println("[+]", "try to connect host:["+address1+"] and ["+address2+"]")
  277. var host1, host2 net.Conn
  278. var err error
  279. for {
  280. d := net.Dialer{Timeout: time.Duration(c.Timeout)}
  281. host1, err = d.Dial("tcp", address1)
  282. if err == nil {
  283. log.Println("[→]", "connect ["+address1+"] success.")
  284. break
  285. } else {
  286. log.Println("[x]", "connect target address ["+address1+"] faild. retry in ", c.Timeout, " seconds. ")
  287. time.Sleep(time.Duration(c.Timeout) * time.Second)
  288. }
  289. if !c.Running {
  290. return nil
  291. }
  292. }
  293. for {
  294. d := net.Dialer{Timeout: time.Duration(c.Timeout)}
  295. host2, err = d.Dial("tcp", address2)
  296. if err == nil {
  297. log.Println("[→]", "connect ["+address2+"] success.")
  298. break
  299. } else {
  300. log.Println("[x]", "connect target address ["+address2+"] faild. retry in ", c.Timeout, " seconds. ")
  301. time.Sleep(time.Duration(c.Timeout) * time.Second)
  302. }
  303. if !c.Running {
  304. return nil
  305. }
  306. }
  307. go forward(host1, host2, &c.aTobAccumulatedByteTransfer, &c.bToaAccumulatedByteTransfer)
  308. }
  309. return nil
  310. }