From 8f088f7ccac9a3c3cb52b109dfdd8ee999564b9e Mon Sep 17 00:00:00 2001 From: penguinehis Date: Sun, 5 Jul 2026 07:43:24 -0300 Subject: [PATCH] Match xray-core: keep connected XHTTP sessions alive; WebSocket early data Two cross-checks against xray-core found separate CRITICAL divergences that stall real traffic: XHTTP: the native session reaper had a 5-minute idle timeout that xray-core does not have. In the default stream-up/stream-down (auto/H2) mode a long download rides inside the already-open GET/POST and generates no new HTTP requests, so the idle timer fired and tore the tunnel down mid-transfer (YouTube/large downloads stalling after a few minutes). Now mirror hub.go: give the downlink GET 30s to attach, and once connected stop reaping entirely -- the session lives for the life of the GET, cleaned up by handleXHTTPDownload's deferred delete. WebSocket: the server handshake ignored Sec-WebSocket-Protocol, silently dropping 0-RTT early data. Clients configured with ?ed=N put the VLESS/VMess request header there and send no first frame, so the server blocked forever waiting for a header that never arrived -- every ed= WS client stalled. Now decode the base64url early data, deliver it before the first frame, and echo the header back, matching transport/internet/websocket. Also match only the path (ignore the ?ed= query) when validating the WS path. Add TestVLESSOverWebSocketEarlyData. Co-Authored-By: Claude Opus 4.8 (1M context) --- xray_native.go | 33 ++++++++++++++++-- xray_native_test.go | 80 +++++++++++++++++++++++++++++++++++++++++++ xray_xhttp.go | 82 ++++++++++++++++++++++----------------------- 3 files changed, 151 insertions(+), 44 deletions(-) diff --git a/xray_native.go b/xray_native.go index 6474bd4..6ae5768 100644 --- a/xray_native.go +++ b/xray_native.go @@ -734,20 +734,43 @@ func wsServerHandshake(conn net.Conn, wantPath string) (*websocketConn, error) { if key == "" { return nil, errors.New("missing Sec-WebSocket-Key") } + // Match only the path portion; the configured path may carry an ?ed=N query + // (early-data hint) and the request path is already query-stripped by Go. + if i := strings.IndexByte(wantPath, '?'); i >= 0 { + wantPath = wantPath[:i] + } if wantPath != "" && wantPath != "/" && req.URL.Path != wantPath { return nil, fmt.Errorf("ws path mismatch: got %q want %q", req.URL.Path, wantPath) } + // WebSocket early data (0-RTT): xray clients configured with ?ed=N carry the + // first bytes -- the VLESS/VMess request header -- base64url-encoded in the + // Sec-WebSocket-Protocol request header instead of a first frame. Decode it and + // feed it to Read before any frame, and echo the header back so the client's + // upgrade check is satisfied. Dropping this makes every ed= WS client stall, + // because the server would wait for a header that never arrives as a frame. + var early []byte + proto := req.Header.Get("Sec-WebSocket-Protocol") + if proto != "" { + if ed, derr := base64.RawURLEncoding.DecodeString(proto); derr == nil { + early = ed + } + } + sum := sha1.Sum([]byte(key + wsMagicGUID)) accept := base64.StdEncoding.EncodeToString(sum[:]) resp := "HTTP/1.1 101 Switching Protocols\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + - "Sec-WebSocket-Accept: " + accept + "\r\n\r\n" + "Sec-WebSocket-Accept: " + accept + "\r\n" + if proto != "" { + resp += "Sec-WebSocket-Protocol: " + proto + "\r\n" + } + resp += "\r\n" if _, err := conn.Write([]byte(resp)); err != nil { return nil, err } - return &websocketConn{Conn: conn, r: br}, nil + return &websocketConn{Conn: conn, r: br, early: early}, nil } // websocketConn adapts a WebSocket data stream to a net.Conn. Client frames are @@ -755,11 +778,17 @@ func wsServerHandshake(conn net.Conn, wantPath string) (*websocketConn, error) { type websocketConn struct { net.Conn r *bufio.Reader + early []byte // 0-RTT early data delivered before the first frame readBuf []byte // decoded payload not yet consumed by Read wmu sync.Mutex } func (c *websocketConn) Read(p []byte) (int, error) { + if len(c.early) > 0 { + n := copy(p, c.early) + c.early = c.early[n:] + return n, nil + } for len(c.readBuf) == 0 { payload, opcode, err := c.readFrame() if err != nil { diff --git a/xray_native_test.go b/xray_native_test.go index 35f23f7..a6059c8 100644 --- a/xray_native_test.go +++ b/xray_native_test.go @@ -238,6 +238,86 @@ func TestVLESSOverWebSocket(t *testing.T) { } } +// TestVLESSOverWebSocketEarlyData guards the 0-RTT path: xray clients configured +// with ?ed=N carry the VLESS request header base64url-encoded in the +// Sec-WebSocket-Protocol header and send no first frame. Dropping it (as the +// handshake did before) makes the server block waiting for a header that never +// arrives as a frame, stalling every ed= WebSocket client. +func TestVLESSOverWebSocketEarlyData(t *testing.T) { + echoPort, stopEcho := startEchoServer(t) + defer stopEcho() + _, port, id, stop := newTestInbound(t, "ws", "/vlws") + defer stop() + + raw, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer raw.Close() + raw.SetDeadline(time.Now().Add(5 * time.Second)) + + // Entire VLESS header + first payload live in the early-data header; no frame. + early := append(vlessHeader(id, echoPort), []byte("ping-ed")...) + proto := base64.RawURLEncoding.EncodeToString(early) + + var keyBytes [16]byte + rand.Read(keyBytes[:]) + key := base64.StdEncoding.EncodeToString(keyBytes[:]) + req := "GET /vlws HTTP/1.1\r\n" + + "Host: test\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Key: " + key + "\r\n" + + "Sec-WebSocket-Protocol: " + proto + "\r\n" + + "Sec-WebSocket-Version: 13\r\n\r\n" + if _, err := raw.Write([]byte(req)); err != nil { + t.Fatalf("handshake write: %v", err) + } + + br := bufio.NewReader(raw) + statusLine, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read status: %v", err) + } + if !strings.Contains(statusLine, "101") { + t.Fatalf("ws handshake not 101: %q", statusLine) + } + sawProto := false + for { + line, err := br.ReadString('\n') + if err != nil { + t.Fatalf("read headers: %v", err) + } + if strings.Contains(line, proto) { + sawProto = true + } + if line == "\r\n" { + break + } + } + if !sawProto { + t.Fatalf("server did not echo Sec-WebSocket-Protocol") + } + + ws := &testWSConn{Conn: raw, r: br} + buf := make([]byte, 0, 32) + want := 2 + len("ping-ed") + for len(buf) < want { + chunk := make([]byte, 64) + n, err := ws.Read(chunk) + if err != nil { + t.Fatalf("ws read (early data dropped?): %v (got %q)", err, buf) + } + buf = append(buf, chunk[:n]...) + } + if buf[0] != 0 { + t.Fatalf("bad ws vless response: %v", buf[:2]) + } + if string(buf[2:want]) != "ping-ed" { + t.Fatalf("ws early-data echo mismatch: got %q", buf[2:want]) + } +} + func TestVLESSOverXHTTPPacketUp(t *testing.T) { echoPort, stopEcho := startEchoServer(t) defer stopEcho() diff --git a/xray_xhttp.go b/xray_xhttp.go index 78de84a..49a917c 100644 --- a/xray_xhttp.go +++ b/xray_xhttp.go @@ -400,10 +400,11 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n xrayTracef("native xray: xhttp session soft limit exceeded inbound=%q active=%d limit=%d", ib.tag, len(ib.xhttpSessions), max) } s := &nativeXHTTPSession{ - id: id, - queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts), - done: make(chan struct{}), - lastSeen: time.Now(), + id: id, + queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts), + done: make(chan struct{}), + connectedCh: make(chan struct{}), + lastSeen: time.Now(), } ib.xhttpSessions[id] = s xrayTracef("native xray: xhttp session created inbound=%q session=%q active=%d", ib.tag, id, len(ib.xhttpSessions)) @@ -419,35 +420,29 @@ func (ib *nativeInbound) xhttpMaxActiveSessions() int { } func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) { - // Keep the cheap unconnected cleanup, but also reap stale sessions that never - // receive their paired download/close because a mobile network or CDN path died. - unconnected := time.NewTimer(20 * time.Second) - stale := time.NewTicker(30 * time.Second) - defer unconnected.Stop() - defer stale.Stop() - for { - select { - case <-unconnected.C: - s.mu.Lock() - connected := s.connected - s.mu.Unlock() - if !connected { - ib.deleteXHTTPSession(id, s) - s.close() - return - } - case <-stale.C: - s.mu.Lock() - idle := time.Since(s.lastSeen) - s.mu.Unlock() - if idle > 5*time.Minute { - ib.deleteXHTTPSession(id, s) - s.close() - return - } - case <-s.done: - return + // Match xray-core hub.go: give the session up to 30s for its downlink GET to + // attach. Once connected, the session lives for the life of that GET request + // with NO idle timeout -- the reaper simply exits. There is deliberately no + // stale/idle reaper for connected sessions: a long stream-down download rides + // inside the already-open GET/POST and produces no new HTTP requests, so an + // idle timeout would kill an active transfer mid-stream (e.g. YouTube stalling + // after a few minutes). Lifetime cleanup is handled by handleXHTTPDownload's + // deferred deleteXHTTPSession when the tunnel ends or the client disconnects. + timer := time.NewTimer(30 * time.Second) + defer timer.Stop() + select { + case <-timer.C: + s.mu.Lock() + connected := s.connected + s.mu.Unlock() + if !connected { + ib.deleteXHTTPSession(id, s) + s.close() } + case <-s.connectedCh: + // Downlink attached in time; stop watching. + case <-s.done: + // Already torn down. } } @@ -712,13 +707,15 @@ func flushHTTP(w http.ResponseWriter) { } type nativeXHTTPSession struct { - id string - queue *nativeXHTTPUploadQueue - done chan struct{} - closeOnce sync.Once - mu sync.Mutex - connected bool - lastSeen time.Time + id string + queue *nativeXHTTPUploadQueue + done chan struct{} + closeOnce sync.Once + connectedCh chan struct{} // closed once the download GET attaches + connectOnce sync.Once + mu sync.Mutex + connected bool + lastSeen time.Time } func (s *nativeXHTTPSession) touch() { @@ -727,14 +724,15 @@ func (s *nativeXHTTPSession) touch() { s.mu.Unlock() } -// markConnected records that the download (stream-down) GET has attached, which -// stops the unconnected-session reaper from expiring it. It does not reject a -// second attach (xray-core does not either): rejecting breaks client reconnects. +// markConnected records that the download (stream-down) GET has attached and +// signals the reaper to stop watching. It does not reject a second attach +// (xray-core does not either): rejecting breaks client reconnects. func (s *nativeXHTTPSession) markConnected() { s.mu.Lock() s.connected = true s.lastSeen = time.Now() s.mu.Unlock() + s.connectOnce.Do(func() { close(s.connectedCh) }) } func (s *nativeXHTTPSession) close() {