hub.go 2.8 KB

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