Fix v2ray HTTP/1.1 transport compatibility

transport/v2rayhttp/client.go

@@ -1,7 +1,6 @@
 package v2rayhttp
 
 import (
-	"bufio"
 	"context"
 	"io"
 	"math/rand"
@@ -81,8 +80,8 @@ func NewClient(ctx context.Context, dialer N.Dialer, serverAddr M.Socksaddr, opt
 	if !strings.HasPrefix(uri.Path, "/") {
 		uri.Path = "/" + uri.Path
 	}
-	for key, value := range options.Headers {
-		client.headers.Set(key, value)
+	for key, valueList := range options.Headers {
+		client.headers[key] = valueList
 	}
 	client.url = &uri
 	return client
@@ -97,18 +96,16 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) {
 }
 
 func (c *Client) dialHTTP(ctx context.Context) (net.Conn, error) {
-	conn, err := c.dialer.DialContext(c.ctx, N.NetworkTCP, c.serverAddr)
+	conn, err := c.dialer.DialContext(ctx, N.NetworkTCP, c.serverAddr)
 	if err != nil {
 		return nil, err
 	}
+
 	request := &http.Request{
-		Method:     c.method,
-		URL:        c.url,
-		ProtoMajor: 1,
-		Proto:      "HTTP/1.1",
-		Header:     c.headers.Clone(),
+		Method: c.method,
+		URL:    c.url,
+		Header: c.headers.Clone(),
 	}
-	request = request.WithContext(ctx)
 	switch hostLen := len(c.host); hostLen {
 	case 0:
 		request.Host = c.serverAddr.AddrString()
@@ -117,30 +114,17 @@ func (c *Client) dialHTTP(ctx context.Context) (net.Conn, error) {
 	default:
 		request.Host = c.host[rand.Intn(hostLen)]
 	}
-	err = request.Write(conn)
-	if err != nil {
-		return nil, err
-	}
-	reader := bufio.NewReader(conn)
-	response, err := http.ReadResponse(reader, request)
-	if err != nil {
-		return nil, err
-	}
-	if response.StatusCode != 200 {
-		return nil, E.New("unexpected status: ", response.Status)
-	}
-	return conn, nil
+
+	return NewHTTP1Conn(conn, request), nil
 }
 
 func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) {
 	pipeInReader, pipeInWriter := io.Pipe()
 	request := &http.Request{
-		Method:     c.method,
-		Body:       pipeInReader,
-		URL:        c.url,
-		ProtoMajor: 2,
-		Proto:      "HTTP/2",
-		Header:     c.headers.Clone(),
+		Method: c.method,
+		Body:   pipeInReader,
+		URL:    c.url,
+		Header: c.headers.Clone(),
 	}
 	request = request.WithContext(ctx)
 	switch hostLen := len(c.host); hostLen {
@@ -152,8 +136,6 @@ func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) {
 	default:
 		request.Host = c.host[rand.Intn(hostLen)]
 	}
-	// Disable any compression method from server.
-	request.Header.Set("Accept-Encoding", "identity")
 	conn := newLateHTTPConn(pipeInWriter)
 	go func() {
 		response, err := c.transport.RoundTrip(request)

transport/v2rayhttp/conn.go

@@ -1,10 +1,12 @@
 package v2rayhttp
 
 import (
+	std_bufio "bufio"
 	"io"
 	"net"
 	"net/http"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -12,37 +14,146 @@ import (
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/buf"
 	"github.com/sagernet/sing/common/bufio"
+	E "github.com/sagernet/sing/common/exceptions"
+	F "github.com/sagernet/sing/common/format"
 	N "github.com/sagernet/sing/common/network"
 )
 
 type HTTPConn struct {
+	net.Conn
+	request        *http.Request
+	requestWritten bool
+	responseRead   bool
+	responseCache  *buf.Buffer
+}
+
+func NewHTTP1Conn(conn net.Conn, request *http.Request) *HTTPConn {
+	return &HTTPConn{
+		Conn:    conn,
+		request: request,
+	}
+}
+
+func (c *HTTPConn) Read(b []byte) (n int, err error) {
+	if !c.responseRead {
+		reader := std_bufio.NewReader(c.Conn)
+		response, err := http.ReadResponse(reader, c.request)
+		if err != nil {
+			return 0, E.Cause(err, "read response")
+		}
+		if response.StatusCode != 200 {
+			return 0, E.New("unexpected status: ", response.Status)
+		}
+		if cacheLen := reader.Buffered(); cacheLen > 0 {
+			c.responseCache = buf.NewSize(cacheLen)
+			_, err = c.responseCache.ReadFullFrom(reader, cacheLen)
+			if err != nil {
+				c.responseCache.Release()
+				return 0, E.Cause(err, "read cache")
+			}
+		}
+		c.responseRead = true
+	}
+	if c.responseCache != nil {
+		n, err = c.responseCache.Read(b)
+		if err == io.EOF {
+			c.responseCache.Release()
+			c.responseCache = nil
+		}
+		if n > 0 {
+			return n, nil
+		}
+	}
+	return c.Conn.Read(b)
+}
+
+func (c *HTTPConn) Write(b []byte) (int, error) {
+	if !c.requestWritten {
+		err := c.writeRequest(b)
+		if err != nil {
+			return 0, E.Cause(err, "write request")
+		}
+		c.requestWritten = true
+		return len(b), nil
+	}
+	return c.Conn.Write(b)
+}
+
+func (c *HTTPConn) writeRequest(payload []byte) error {
+	writer := bufio.NewBufferedWriter(c.Conn, buf.New())
+	const CRLF = "\r\n"
+	_, err := writer.Write([]byte(F.ToString(c.request.Method, " ", c.request.URL.RequestURI(), " HTTP/1.1", CRLF)))
+	if err != nil {
+		return err
+	}
+	if c.request.Header.Get("Host") == "" {
+		c.request.Header.Set("Host", c.request.Host)
+	}
+	for key, value := range c.request.Header {
+		_, err = writer.Write([]byte(F.ToString(key, ": ", strings.Join(value, ", "), CRLF)))
+		if err != nil {
+			return err
+		}
+	}
+	_, err = writer.Write([]byte(CRLF))
+	if err != nil {
+		return err
+	}
+	_, err = writer.Write(payload)
+	if err != nil {
+		return err
+	}
+	err = writer.Fallthrough()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *HTTPConn) ReaderReplaceable() bool {
+	return c.responseRead
+}
+
+func (c *HTTPConn) WriterReplaceable() bool {
+	return c.requestWritten
+}
+
+func (c *HTTPConn) NeedHandshake() bool {
+	return !c.requestWritten
+}
+
+func (c *HTTPConn) Upstream() any {
+	return c.Conn
+}
+
+type HTTP2Conn struct {
 	reader io.Reader
 	writer io.Writer
 	create chan struct{}
 	err    error
 }
 
-func NewHTTPConn(reader io.Reader, writer io.Writer) HTTPConn {
-	return HTTPConn{
+func NewHTTPConn(reader io.Reader, writer io.Writer) HTTP2Conn {
+	return HTTP2Conn{
 		reader: reader,
 		writer: writer,
 	}
 }
 
-func newLateHTTPConn(writer io.Writer) *HTTPConn {
-	return &HTTPConn{
+func newLateHTTPConn(writer io.Writer) *HTTP2Conn {
+	return &HTTP2Conn{
 		create: make(chan struct{}),
 		writer: writer,
 	}
 }
 
-func (c *HTTPConn) setup(reader io.Reader, err error) {
+func (c *HTTP2Conn) setup(reader io.Reader, err error) {
 	c.reader = reader
 	c.err = err
 	close(c.create)
 }
 
-func (c *HTTPConn) Read(b []byte) (n int, err error) {
+func (c *HTTP2Conn) Read(b []byte) (n int, err error) {
 	if c.reader == nil {
 		<-c.create
 		if c.err != nil {
@@ -53,24 +164,24 @@ func (c *HTTPConn) Read(b []byte) (n int, err error) {
 	return n, baderror.WrapH2(err)
 }
 
-func (c *HTTPConn) Write(b []byte) (n int, err error) {
+func (c *HTTP2Conn) Write(b []byte) (n int, err error) {
 	n, err = c.writer.Write(b)
 	return n, baderror.WrapH2(err)
 }
 
-func (c *HTTPConn) Close() error {
+func (c *HTTP2Conn) Close() error {
 	return common.Close(c.reader, c.writer)
 }
 
-func (c *HTTPConn) LocalAddr() net.Addr {
+func (c *HTTP2Conn) LocalAddr() net.Addr {
 	return nil
 }
 
-func (c *HTTPConn) RemoteAddr() net.Addr {
+func (c *HTTP2Conn) RemoteAddr() net.Addr {
 	return nil
 }
 
-func (c *HTTPConn) SetDeadline(t time.Time) error {
+func (c *HTTP2Conn) SetDeadline(t time.Time) error {
 	if responseWriter, loaded := c.writer.(interface {
 		SetWriteDeadline(time.Time) error
 	}); loaded {
@@ -79,7 +190,7 @@ func (c *HTTPConn) SetDeadline(t time.Time) error {
 	return os.ErrInvalid
 }
 
-func (c *HTTPConn) SetReadDeadline(t time.Time) error {
+func (c *HTTP2Conn) SetReadDeadline(t time.Time) error {
 	if responseWriter, loaded := c.writer.(interface {
 		SetReadDeadline(time.Time) error
 	}); loaded {
@@ -88,7 +199,7 @@ func (c *HTTPConn) SetReadDeadline(t time.Time) error {
 	return os.ErrInvalid
 }
 
-func (c *HTTPConn) SetWriteDeadline(t time.Time) error {
+func (c *HTTP2Conn) SetWriteDeadline(t time.Time) error {
 	if responseWriter, loaded := c.writer.(interface {
 		SetWriteDeadline(time.Time) error
 	}); loaded {
@@ -98,7 +209,7 @@ func (c *HTTPConn) SetWriteDeadline(t time.Time) error {
 }
 
 type ServerHTTPConn struct {
-	HTTPConn
+	HTTP2Conn
 	flusher http.Flusher
 }
 

transport/v2rayhttp/server.go

@@ -13,6 +13,8 @@ import (
 	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/option"
 	"github.com/sagernet/sing/common"
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
 	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
@@ -62,7 +64,7 @@ func NewServer(ctx context.Context, options option.V2RayHTTPOptions, tlsConfig t
 		server.path = "/" + server.path
 	}
 	for key, value := range options.Headers {
-		server.headers.Set(key, value)
+		server.headers[key] = value
 	}
 	server.httpServer = &http.Server{
 		Handler:           server,
@@ -106,11 +108,20 @@ func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	var metadata M.Metadata
 	metadata.Source = sHttp.SourceAddress(request)
 	if h, ok := writer.(http.Hijacker); ok {
-		conn, _, err := h.Hijack()
+		conn, reader, err := h.Hijack()
 		if err != nil {
 			s.fallbackRequest(request.Context(), writer, request, http.StatusInternalServerError, E.Cause(err, "hijack conn"))
 			return
 		}
+		if cacheLen := reader.Reader.Buffered(); cacheLen > 0 {
+			cache := buf.NewSize(cacheLen)
+			_, err = cache.ReadFullFrom(reader.Reader, cacheLen)
+			if err != nil {
+				s.fallbackRequest(request.Context(), writer, request, http.StatusInternalServerError, E.Cause(err, "read cache"))
+				return
+			}
+			conn = bufio.NewCachedConn(conn, cache)
+		}
 		s.handler.NewConnection(request.Context(), conn, metadata)
 	} else {
 		conn := NewHTTP2Wrapper(&ServerHTTPConn{