mirror of
https://github.com/XTLS/Xray-core.git
synced 2026-10-21 16:54:29 +00:00
HTTP outbound: Fix hanging on server-first traffic (#7193)
https://github.com/XTLS/Xray-core/pull/7193#issuecomment-6312885274 Fixes https://github.com/XTLS/Xray-core/issues/3459 --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: 风扇滑翔翼 <Fangliding.fshxy@outlook.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
风扇滑翔翼
parent
23e2292344
commit
061c2df291
+30
-2
@@ -10,6 +10,7 @@ import (
|
||||
"net/url"
|
||||
"sync"
|
||||
"text/template"
|
||||
"time"
|
||||
|
||||
"github.com/xtls/xray-core/common"
|
||||
"github.com/xtls/xray-core/common/buf"
|
||||
@@ -86,7 +87,13 @@ func (c *Client) Process(ctx context.Context, link *transport.Link, dialer inter
|
||||
user := server.User
|
||||
var conn stat.Connection
|
||||
|
||||
mbuf, _ := link.Reader.ReadMultiBuffer()
|
||||
var mbuf buf.MultiBuffer
|
||||
if reader, ok := link.Reader.(buf.TimeoutReader); ok {
|
||||
// Don't wait forever: server-first protocols (SMTP, FTP, MySQL...) send nothing until the server greets.
|
||||
mbuf, _ = reader.ReadMultiBufferTimeout(time.Millisecond * 500)
|
||||
} else {
|
||||
mbuf, _ = link.Reader.ReadMultiBuffer()
|
||||
}
|
||||
len := mbuf.Len()
|
||||
firstPayload := bytespool.Alloc(len)
|
||||
mbuf, _ = buf.SplitBytes(mbuf, firstPayload)
|
||||
@@ -234,7 +241,8 @@ func setUpHTTPTunnel(ctx context.Context, dest net.Destination, target string, u
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := http.ReadResponse(bufio.NewReader(rawConn), req)
|
||||
reader := bufio.NewReader(rawConn)
|
||||
resp, err := http.ReadResponse(reader, req)
|
||||
if err != nil {
|
||||
rawConn.Close()
|
||||
return nil, err
|
||||
@@ -245,6 +253,11 @@ func setUpHTTPTunnel(ctx context.Context, dest net.Destination, target string, u
|
||||
rawConn.Close()
|
||||
return nil, errors.New("Proxy responded with non 200 code: " + resp.Status)
|
||||
}
|
||||
// Tunneled bytes (e.g. a server-first greeting) may arrive in the same read as the 200 response.
|
||||
if n := reader.Buffered(); n > 0 {
|
||||
early, _ := reader.Peek(n)
|
||||
return &earlyDataConn{Conn: rawConn, early: early}, nil
|
||||
}
|
||||
return rawConn, nil
|
||||
}
|
||||
|
||||
@@ -375,6 +388,21 @@ func (h *http2Conn) Close() error {
|
||||
return h.out.Close()
|
||||
}
|
||||
|
||||
// earlyDataConn returns the bytes read ahead together with the CONNECT response before reading from Conn.
|
||||
type earlyDataConn struct {
|
||||
net.Conn
|
||||
early []byte
|
||||
}
|
||||
|
||||
func (c *earlyDataConn) Read(p []byte) (int, error) {
|
||||
if len(c.early) > 0 {
|
||||
n := copy(p, c.early)
|
||||
c.early = c.early[n:]
|
||||
return n, nil
|
||||
}
|
||||
return c.Conn.Read(p)
|
||||
}
|
||||
|
||||
func init() {
|
||||
common.Must(common.RegisterConfig((*ClientConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
|
||||
return NewClient(ctx, config.(*ClientConfig))
|
||||
|
||||
@@ -16,8 +16,10 @@ import (
|
||||
"github.com/xtls/xray-core/common"
|
||||
"github.com/xtls/xray-core/common/buf"
|
||||
"github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/protocol"
|
||||
"github.com/xtls/xray-core/common/serial"
|
||||
"github.com/xtls/xray-core/core"
|
||||
"github.com/xtls/xray-core/proxy/dokodemo"
|
||||
"github.com/xtls/xray-core/proxy/freedom"
|
||||
v2http "github.com/xtls/xray-core/proxy/http"
|
||||
v2httptest "github.com/xtls/xray-core/testing/servers/http"
|
||||
@@ -377,3 +379,88 @@ func TestHttpBasicAuth(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Server-first protocols (SMTP, FTP, MySQL) send nothing from the client until the server's
|
||||
// greeting arrives, so the HTTP outbound must not wait for a first payload before dialing.
|
||||
func TestHTTPOutboundServerFirst(t *testing.T) {
|
||||
tcpServer := tcp.Server{
|
||||
MsgProcessor: xor,
|
||||
SendFirst: []byte("send first"),
|
||||
}
|
||||
dest, err := tcpServer.Start()
|
||||
common.Must(err)
|
||||
defer tcpServer.Close()
|
||||
|
||||
serverPort := tcp.PickPort()
|
||||
serverConfig := &core.Config{
|
||||
Inbound: []*core.InboundHandlerConfig{
|
||||
{
|
||||
ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{
|
||||
PortList: &net.PortList{Range: []*net.PortRange{net.SinglePortRange(serverPort)}},
|
||||
Listen: net.NewIPOrDomain(net.LocalHostIP),
|
||||
}),
|
||||
ProxySettings: serial.ToTypedMessage(&v2http.ServerConfig{}),
|
||||
},
|
||||
},
|
||||
Outbound: []*core.OutboundHandlerConfig{
|
||||
{
|
||||
ProxySettings: serial.ToTypedMessage(&freedom.Config{
|
||||
FinalRules: []*freedom.FinalRuleConfig{{Action: freedom.RuleAction_Allow}},
|
||||
}),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
clientPort := tcp.PickPort()
|
||||
clientConfig := &core.Config{
|
||||
Inbound: []*core.InboundHandlerConfig{
|
||||
{
|
||||
ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{
|
||||
PortList: &net.PortList{Range: []*net.PortRange{net.SinglePortRange(clientPort)}},
|
||||
Listen: net.NewIPOrDomain(net.LocalHostIP),
|
||||
}),
|
||||
ProxySettings: serial.ToTypedMessage(&dokodemo.Config{
|
||||
RewriteAddress: net.NewIPOrDomain(dest.Address),
|
||||
RewritePort: uint32(dest.Port),
|
||||
AllowedNetworks: []net.Network{net.Network_TCP},
|
||||
}),
|
||||
},
|
||||
},
|
||||
Outbound: []*core.OutboundHandlerConfig{
|
||||
{
|
||||
ProxySettings: serial.ToTypedMessage(&v2http.ClientConfig{
|
||||
Server: &protocol.ServerEndpoint{
|
||||
Address: net.NewIPOrDomain(net.LocalHostIP),
|
||||
Port: uint32(serverPort),
|
||||
},
|
||||
}),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
servers, err := InitializeServerConfigs(serverConfig, clientConfig)
|
||||
common.Must(err)
|
||||
defer CloseAllServers(servers)
|
||||
|
||||
conn, err := net.DialTCP("tcp", nil, &net.TCPAddr{
|
||||
IP: []byte{127, 0, 0, 1},
|
||||
Port: int(clientPort),
|
||||
})
|
||||
common.Must(err)
|
||||
defer conn.Close()
|
||||
|
||||
common.Must(conn.SetReadDeadline(time.Now().Add(5 * time.Second)))
|
||||
response := make([]byte, 1024)
|
||||
nBytes, err := conn.Read(response)
|
||||
if err != nil {
|
||||
t.Fatal("no server greeting through HTTP outbound: ", err)
|
||||
}
|
||||
if string(response[:nBytes]) != "send first" {
|
||||
t.Fatal("unexpected first response: ", string(response[:nBytes]))
|
||||
}
|
||||
common.Must(conn.SetReadDeadline(time.Time{}))
|
||||
|
||||
if err := testTCPConn2(conn, 1024, time.Second*5)(); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user