hub.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  1. // +build !confonly
  2. package http
  3. import (
  4. "context"
  5. "io"
  6. "net/http"
  7. "strings"
  8. "time"
  9. "v2ray.com/core/common"
  10. "v2ray.com/core/common/net"
  11. "v2ray.com/core/common/serial"
  12. "v2ray.com/core/common/session"
  13. "v2ray.com/core/common/signal/done"
  14. "v2ray.com/core/transport/internet"
  15. "v2ray.com/core/transport/internet/tls"
  16. "golang.org/x/net/http2"
  17. "golang.org/x/net/http2/h2c"
  18. )
  19. type Listener struct {
  20. server *http.Server
  21. handler internet.ConnHandler
  22. local net.Addr
  23. config Config
  24. }
  25. func (l *Listener) Addr() net.Addr {
  26. return l.local
  27. }
  28. func (l *Listener) Close() error {
  29. return l.server.Close()
  30. }
  31. type flushWriter struct {
  32. w io.Writer
  33. d *done.Instance
  34. }
  35. func (fw flushWriter) Write(p []byte) (n int, err error) {
  36. if fw.d.Done() {
  37. return 0, io.ErrClosedPipe
  38. }
  39. n, err = fw.w.Write(p)
  40. if f, ok := fw.w.(http.Flusher); ok {
  41. f.Flush()
  42. }
  43. return
  44. }
  45. func (l *Listener) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
  46. host := request.Host
  47. if !l.config.isValidHost(host) {
  48. writer.WriteHeader(404)
  49. return
  50. }
  51. path := l.config.getNormalizedPath()
  52. if !strings.HasPrefix(request.URL.Path, path) {
  53. writer.WriteHeader(404)
  54. return
  55. }
  56. writer.Header().Set("Cache-Control", "no-store")
  57. writer.WriteHeader(200)
  58. if f, ok := writer.(http.Flusher); ok {
  59. f.Flush()
  60. }
  61. remoteAddr := l.Addr()
  62. dest, err := net.ParseDestination(request.RemoteAddr)
  63. if err != nil {
  64. newError("failed to parse request remote addr: ", request.RemoteAddr).Base(err).WriteToLog()
  65. } else {
  66. remoteAddr = &net.TCPAddr{
  67. IP: dest.Address.IP(),
  68. Port: int(dest.Port),
  69. }
  70. }
  71. done := done.New()
  72. conn := net.NewConnection(
  73. net.ConnectionOutput(request.Body),
  74. net.ConnectionInput(flushWriter{w: writer, d: done}),
  75. net.ConnectionOnClose(common.ChainedClosable{done, request.Body}),
  76. net.ConnectionLocalAddr(l.Addr()),
  77. net.ConnectionRemoteAddr(remoteAddr),
  78. )
  79. l.handler(conn)
  80. <-done.Wait()
  81. }
  82. func Listen(ctx context.Context, address net.Address, port net.Port, streamSettings *internet.MemoryStreamConfig, handler internet.ConnHandler) (internet.Listener, error) {
  83. httpSettings := streamSettings.ProtocolSettings.(*Config)
  84. listener := &Listener{
  85. handler: handler,
  86. local: &net.TCPAddr{
  87. IP: address.IP(),
  88. Port: int(port),
  89. },
  90. config: *httpSettings,
  91. }
  92. config := tls.ConfigFromStreamSettings(streamSettings)
  93. if config == nil {
  94. // return nil, newError("TLS must be enabled for http transport.").AtWarning()
  95. h2s:=&http2.Server{}
  96. server := &http.Server{
  97. Addr: serial.Concat(address, ":", port),
  98. // TLSConfig: config.GetTLSConfig(tls.WithNextProto("h2")),
  99. Handler: h2c.NewHandler(listener,h2s),
  100. ReadHeaderTimeout: time.Second * 4,
  101. }
  102. listener.server = server
  103. go func() {
  104. tcpListener, err := internet.ListenSystem(ctx, &net.TCPAddr{
  105. IP: address.IP(),
  106. Port: int(port),
  107. }, streamSettings.SocketSettings)
  108. if err != nil {
  109. newError("failed to listen on", address, ":", port).Base(err).WriteToLog(session.ExportIDToError(ctx))
  110. return
  111. }
  112. err = server.Serve(tcpListener)
  113. if err != nil {
  114. newError("stoping serving H2C").Base(err).WriteToLog(session.ExportIDToError(ctx))
  115. }
  116. }()
  117. return listener, nil
  118. } else {
  119. server := &http.Server{
  120. Addr: serial.Concat(address, ":", port),
  121. TLSConfig: config.GetTLSConfig(tls.WithNextProto("h2")),
  122. Handler: listener,
  123. ReadHeaderTimeout: time.Second * 4,
  124. }
  125. listener.server = server
  126. go func() {
  127. tcpListener, err := internet.ListenSystem(ctx, &net.TCPAddr{
  128. IP: address.IP(),
  129. Port: int(port),
  130. }, streamSettings.SocketSettings)
  131. if err != nil {
  132. newError("failed to listen on", address, ":", port).Base(err).WriteToLog(session.ExportIDToError(ctx))
  133. return
  134. }
  135. err = server.ServeTLS(tcpListener, "", "")
  136. if err != nil {
  137. newError("stoping serving TLS").Base(err).WriteToLog(session.ExportIDToError(ctx))
  138. }
  139. }()
  140. return listener, nil
  141. }
  142. }
  143. func init() {
  144. common.Must(internet.RegisterTransportListener(protocolName, Listen))
  145. }