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) <noreply@anthropic.com>
2026-07-05 07:43:24 -03:00
parent 7200dbd236
commit 8f088f7cca
3 changed files with 151 additions and 44 deletions
@@ -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 {
@@ -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()
@@ -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() {