hub.go 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. package http
  2. import (
  3. "context"
  4. "io"
  5. "net/http"
  6. "strings"
  7. "v2ray.com/core/common"
  8. "v2ray.com/core/common/net"
  9. "v2ray.com/core/common/serial"
  10. "v2ray.com/core/common/session"
  11. "v2ray.com/core/common/signal/done"
  12. "v2ray.com/core/transport/internet"
  13. "v2ray.com/core/transport/internet/tls"
  14. )
  15. type Listener struct {
  16. server *http.Server
  17. handler internet.ConnHandler
  18. local net.Addr
  19. config Config
  20. }
  21. func (l *Listener) Addr() net.Addr {
  22. return l.local
  23. }
  24. func (l *Listener) Close() error {
  25. return l.server.Close()
  26. }
  27. type flushWriter struct {
  28. w io.Writer
  29. d *done.Instance
  30. }
  31. func (fw flushWriter) Write(p []byte) (n int, err error) {
  32. if fw.d.Done() {
  33. return 0, io.ErrClosedPipe
  34. }
  35. n, err = fw.w.Write(p)
  36. if f, ok := fw.w.(http.Flusher); ok {
  37. f.Flush()
  38. }
  39. return
  40. }
  41. func (l *Listener) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
  42. host := request.Host
  43. if !l.config.isValidHost(host) {
  44. writer.WriteHeader(404)
  45. return
  46. }
  47. path := l.config.getNormalizedPath()
  48. if !strings.HasPrefix(request.URL.Path, path) {
  49. writer.WriteHeader(404)
  50. return
  51. }
  52. writer.Header().Set("Cache-Control", "no-store")
  53. writer.WriteHeader(200)
  54. if f, ok := writer.(http.Flusher); ok {
  55. f.Flush()
  56. }
  57. remoteAddr := l.Addr()
  58. dest, err := net.ParseDestination(request.RemoteAddr)
  59. if err != nil {
  60. newError("failed to parse request remote addr: ", request.RemoteAddr).Base(err).WriteToLog()
  61. } else {
  62. remoteAddr = &net.TCPAddr{
  63. IP: dest.Address.IP(),
  64. Port: int(dest.Port),
  65. }
  66. }
  67. done := done.New()
  68. conn := net.NewConnection(
  69. net.ConnectionOutput(request.Body),
  70. net.ConnectionInput(flushWriter{w: writer, d: done}),
  71. net.ConnectionOnClose(common.NewChainedClosable(done, request.Body)),
  72. net.ConnectionLocalAddr(l.Addr()),
  73. net.ConnectionRemoteAddr(remoteAddr),
  74. )
  75. l.handler(conn)
  76. <-done.Wait()
  77. }
  78. func Listen(ctx context.Context, address net.Address, port net.Port, handler internet.ConnHandler) (internet.Listener, error) {
  79. rawSettings := internet.StreamSettingsFromContext(ctx)
  80. httpSettings, ok := rawSettings.ProtocolSettings.(*Config)
  81. if !ok {
  82. return nil, newError("HTTP config is not set.").AtError()
  83. }
  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.ConfigFromContext(ctx)
  93. if config == nil {
  94. return nil, newError("TLS must be enabled for http transport.").AtWarning()
  95. }
  96. server := &http.Server{
  97. Addr: serial.Concat(address, ":", port),
  98. TLSConfig: config.GetTLSConfig(tls.WithNextProto("h2")),
  99. Handler: listener,
  100. }
  101. listener.server = server
  102. go func() {
  103. tcpListener, err := internet.ListenSystem(ctx, &net.TCPAddr{
  104. IP: address.IP(),
  105. Port: int(port),
  106. })
  107. if err != nil {
  108. newError("failed to listen on", address, ":", port).Base(err).WriteToLog(session.ExportIDToError(ctx))
  109. return
  110. }
  111. err = server.ServeTLS(tcpListener, "", "")
  112. if err != nil {
  113. newError("stoping serving TLS").Base(err).WriteToLog(session.ExportIDToError(ctx))
  114. }
  115. }()
  116. return listener, nil
  117. }
  118. func init() {
  119. common.Must(internet.RegisterTransportListener(protocolName, Listen))
  120. }