hub.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. // +build !confonly
  2. package websocket
  3. import (
  4. "context"
  5. "crypto/tls"
  6. "net/http"
  7. "sync"
  8. "time"
  9. "github.com/gorilla/websocket"
  10. "github.com/v2fly/v2ray-core/v4/common"
  11. "github.com/v2fly/v2ray-core/v4/common/net"
  12. http_proto "github.com/v2fly/v2ray-core/v4/common/protocol/http"
  13. "github.com/v2fly/v2ray-core/v4/common/session"
  14. "github.com/v2fly/v2ray-core/v4/transport/internet"
  15. v2tls "github.com/v2fly/v2ray-core/v4/transport/internet/tls"
  16. )
  17. type requestHandler struct {
  18. path string
  19. ln *Listener
  20. }
  21. var upgrader = &websocket.Upgrader{
  22. ReadBufferSize: 4 * 1024,
  23. WriteBufferSize: 4 * 1024,
  24. HandshakeTimeout: time.Second * 4,
  25. CheckOrigin: func(r *http.Request) bool {
  26. return true
  27. },
  28. }
  29. func (h *requestHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
  30. if request.URL.Path != h.path {
  31. writer.WriteHeader(http.StatusNotFound)
  32. return
  33. }
  34. conn, err := upgrader.Upgrade(writer, request, nil)
  35. if err != nil {
  36. newError("failed to convert to WebSocket connection").Base(err).WriteToLog()
  37. return
  38. }
  39. forwardedAddrs := http_proto.ParseXForwardedFor(request.Header)
  40. remoteAddr := conn.RemoteAddr()
  41. if len(forwardedAddrs) > 0 && forwardedAddrs[0].Family().IsIP() {
  42. remoteAddr = &net.TCPAddr{
  43. IP: forwardedAddrs[0].IP(),
  44. Port: int(0),
  45. }
  46. }
  47. h.ln.addConn(newConnection(conn, remoteAddr))
  48. }
  49. type Listener struct {
  50. sync.Mutex
  51. server http.Server
  52. listener net.Listener
  53. config *Config
  54. addConn internet.ConnHandler
  55. locker *internet.FileLocker // for unix domain socket
  56. }
  57. func ListenWS(ctx context.Context, address net.Address, port net.Port, streamSettings *internet.MemoryStreamConfig, addConn internet.ConnHandler) (internet.Listener, error) {
  58. l := &Listener{
  59. addConn: addConn,
  60. }
  61. wsSettings := streamSettings.ProtocolSettings.(*Config)
  62. l.config = wsSettings
  63. if l.config != nil {
  64. if streamSettings.SocketSettings == nil {
  65. streamSettings.SocketSettings = &internet.SocketConfig{}
  66. }
  67. streamSettings.SocketSettings.AcceptProxyProtocol = l.config.AcceptProxyProtocol
  68. }
  69. var listener net.Listener
  70. var err error
  71. if port == net.Port(0) { // unix
  72. listener, err = internet.ListenSystem(ctx, &net.UnixAddr{
  73. Name: address.Domain(),
  74. Net: "unix",
  75. }, streamSettings.SocketSettings)
  76. if err != nil {
  77. return nil, newError("failed to listen unix domain socket(for WS) on ", address).Base(err)
  78. }
  79. newError("listening unix domain socket(for WS) on ", address).WriteToLog(session.ExportIDToError(ctx))
  80. locker := ctx.Value(address.Domain())
  81. if locker != nil {
  82. l.locker = locker.(*internet.FileLocker)
  83. }
  84. } else { // tcp
  85. listener, err = internet.ListenSystem(ctx, &net.TCPAddr{
  86. IP: address.IP(),
  87. Port: int(port),
  88. }, streamSettings.SocketSettings)
  89. if err != nil {
  90. return nil, newError("failed to listen TCP(for WS) on ", address, ":", port).Base(err)
  91. }
  92. newError("listening TCP(for WS) on ", address, ":", port).WriteToLog(session.ExportIDToError(ctx))
  93. }
  94. if streamSettings.SocketSettings != nil && streamSettings.SocketSettings.AcceptProxyProtocol {
  95. newError("accepting PROXY protocol").AtWarning().WriteToLog(session.ExportIDToError(ctx))
  96. }
  97. if config := v2tls.ConfigFromStreamSettings(streamSettings); config != nil {
  98. if tlsConfig := config.GetTLSConfig(); tlsConfig != nil {
  99. listener = tls.NewListener(listener, tlsConfig)
  100. }
  101. }
  102. l.listener = listener
  103. l.server = http.Server{
  104. Handler: &requestHandler{
  105. path: wsSettings.GetNormalizedPath(),
  106. ln: l,
  107. },
  108. ReadHeaderTimeout: time.Second * 4,
  109. MaxHeaderBytes: 2048,
  110. }
  111. go func() {
  112. if err := l.server.Serve(l.listener); err != nil {
  113. newError("failed to serve http for WebSocket").Base(err).AtWarning().WriteToLog(session.ExportIDToError(ctx))
  114. }
  115. }()
  116. return l, err
  117. }
  118. // Addr implements net.Listener.Addr().
  119. func (ln *Listener) Addr() net.Addr {
  120. return ln.listener.Addr()
  121. }
  122. // Close implements net.Listener.Close().
  123. func (ln *Listener) Close() error {
  124. if ln.locker != nil {
  125. ln.locker.Release()
  126. }
  127. return ln.listener.Close()
  128. }
  129. func init() {
  130. common.Must(internet.RegisterTransportListener(protocolName, ListenWS))
  131. }