Try pass RST

https://github.com/XTLS/Xray-core/pull/6640#issuecomment-5420106315

Fixes https://github.com/XTLS/Xray-core/issues/6636
Fixes https://github.com/XTLS/Xray-core/issues/6600
...

app/observatory/observer.go

@@ -78,6 +78,12 @@ func (o *Observer) background() {
 			sleepTime = time.Duration(o.config.ProbeInterval)
 		}
 
+		if len(outbounds) == 0 {
+			errors.LogWarning(o.ctx, "no outbound matches subjectSelector ", o.config.SubjectSelector)
+			time.Sleep(sleepTime)
+			continue
+		}
+
 		if !o.config.EnableConcurrency {
 			sort.Strings(outbounds)
 			for _, v := range outbounds {

common/reflect/marshal_test.go

@@ -207,6 +207,7 @@ func getConfig() string {
       "tag": "XHTTP_IN",
       "streamSettings": {
         "network": "xhttp",
+		"security": "tls",
         "xhttpSettings": {
           "host": "bing.com",
           "path": "/xhttp_client_upload",

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/sagernet/sing-shadowsocks v0.2.7
 	github.com/stretchr/testify v1.12.1
 	github.com/vishvananda/netlink v1.3.1
-	github.com/xtls/reality v0.0.0-20260322125925-9234c772ba8f
+	github.com/xtls/reality v0.0.0-20260827183302-8530a57042be
 	go4.org/netipx v0.0.0-20231129151722-fdeea329fbba
 	golang.org/x/crypto v0.55.0
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842

go.sum

@@ -73,8 +73,8 @@ github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zd
 github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
 github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
 github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
-github.com/xtls/reality v0.0.0-20260322125925-9234c772ba8f h1:iy2JRioxmUpoJ3SzbFPyTxHZMbR/rSHP7dOOgYaq1O8=
+github.com/xtls/reality v0.0.0-20260827183302-8530a57042be h1:0MCMg+ylSR2kIWtRUgMH+1zk+lRksbH7pIY4lGHSFKY=
-github.com/xtls/reality v0.0.0-20260322125925-9234c772ba8f/go.mod h1:DsJblcWDGt76+FVqBVwbwRhxyyNJsGV48gJLch0OOWI=
+github.com/xtls/reality v0.0.0-20260827183302-8530a57042be/go.mod h1:DsJblcWDGt76+FVqBVwbwRhxyyNJsGV48gJLch0OOWI=
 github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=

infra/conf/common.go

@@ -65,6 +65,9 @@ func (v *Address) UnmarshalJSON(data []byte) error {
 }
 
 func (v *Address) Build() *net.IPOrDomain {
+	if v == nil {
+		return nil
+	}
 	return net.NewIPOrDomain(v.Address)
 }
 

infra/conf/xray.go

@@ -140,7 +140,7 @@ func (c *InboundDetourConfig) Build() (*core.InboundHandlerConfig, error) {
 	// TUN inbound doesn't need port configuration as it uses network interface instead
 	if strings.ToLower(c.Protocol) == "tun" {
 		// Skip port validation for TUN
-	} else if c.ListenOn == nil {
+	} else if c.ListenOn == nil || len(c.ListenOn.String()) == 0 {
 		// Listen on anyip, must set PortList
 		if c.PortList == nil {
 			return nil, errors.New("Listen on AnyIP but no Port(s) set in InboundDetour.")
@@ -251,13 +251,13 @@ func validateOutboundTransportSecurity(rawConfig interface{}, senderSettings *pr
 		if vlessCfg.Encryption != "" && vlessCfg.Encryption != "none" {
 			return nil
 		}
-		if requiresTransportSecurity(vlessCfg.Address) {
+		if requiresTransportSecurity(vlessCfg.Vnext[0].Address) {
 			return errors.New("vless without TLS or other encryption is prohibited unless the server address is a private IP or domain")
 		}
 	}
 
 	if tjCfg, ok := rawConfig.(*TrojanClientConfig); ok {
-		if requiresTransportSecurity(tjCfg.Address) {
+		if requiresTransportSecurity(tjCfg.Servers[0].Address) {
 			return errors.New("trojan without TLS is prohibited unless the server address is a private IP or domain")
 		}
 	}
@@ -358,13 +358,13 @@ func (c *OutboundDetourConfig) Build() (*core.OutboundHandlerConfig, error) {
 	if err != nil {
 		return nil, errors.New("failed to load outbound detour config for protocol ", c.Protocol).Base(err)
 	}
-	if err := validateOutboundTransportSecurity(rawConfig, senderSettings); err != nil {
-		return nil, err
-	}
 	ts, err := rawConfig.(Buildable).Build()
 	if err != nil {
 		return nil, errors.New("failed to build outbound handler for protocol ", c.Protocol).Base(err)
 	}
+	if err := validateOutboundTransportSecurity(rawConfig, senderSettings); err != nil {
+		return nil, err
+	}
 
 	return &core.OutboundHandlerConfig{
 		SenderSettings: serial.ToTypedMessage(senderSettings),

proxy/dokodemo/fakeudp_openbsd.go

@@ -12,12 +12,29 @@ import (
 )
 
 func FakeUDP(addr *net.UDPAddr, mark int) (net.PacketConn, error) {
-	ip4 := addr.IP.To4()
+	domain := unix.AF_INET6
-	if ip4 == nil {
+	var sockaddr unix.Sockaddr
-		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("IPv6 is not supported by the OpenBSD transparent UDP patch")}
+	if ip4 := addr.IP.To4(); ip4 != nil {
+		domain = unix.AF_INET
+		sa := &unix.SockaddrInet4{Port: addr.Port}
+		copy(sa.Addr[:], ip4)
+		sockaddr = sa
+	} else if ip6 := addr.IP.To16(); ip6 != nil {
+		sa := &unix.SockaddrInet6{Port: addr.Port}
+		copy(sa.Addr[:], ip6)
+		if addr.Zone != "" {
+			iface, err := net.InterfaceByName(addr.Zone)
+			if err != nil {
+				return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("resolve zone %s: %w", addr.Zone, err)}
+			}
+			sa.ZoneId = uint32(iface.Index)
+		}
+		sockaddr = sa
+	} else {
+		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("unsupported address %v", addr.IP)}
 	}
 
-	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, 0)
+	fd, err := unix.Socket(domain, unix.SOCK_DGRAM, 0)
 	if err != nil {
 		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("socket open: %w", err)}
 	}
@@ -35,9 +52,12 @@ func FakeUDP(addr *net.UDPAddr, mark int) (net.PacketConn, error) {
 	if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEADDR, 1); err != nil {
 		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("set socket option SO_REUSEADDR: %w", err)}
 	}
+	// Several client sessions can be answered from the same original
+	// destination at the same time, so the address has to be shareable.
+	if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1); err != nil {
+		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("set socket option SO_REUSEPORT: %w", err)}
+	}
 
-	sockaddr := &unix.SockaddrInet4{Port: addr.Port}
-	copy(sockaddr.Addr[:], ip4)
 	if err = unix.Bind(fd, sockaddr); err != nil {
 		return nil, &net.OpError{Op: "fake", Err: fmt.Errorf("bind %s: %w", addr.String(), err)}
 	}

proxy/http/client.go

@@ -173,15 +173,18 @@ func fillRequestHeader(ctx context.Context, header []*Header) ([]*Header, error)
 	outbounds := session.OutboundsFromContext(ctx)
 	ob := outbounds[len(outbounds)-1]
 
-	if inbound == nil || ob == nil {
+	var src net.Destination
-		return nil, errors.New("missing inbound or outbound metadata from context")
+	if inbound != nil {
+		src = inbound.Source
+	} else {
+		src = net.TCPDestination(net.AnyIP, 0)
 	}
 
 	data := struct {
 		Source net.Destination
 		Target net.Destination
 	}{
-		Source: inbound.Source,
+		Source: src,
 		Target: ob.Target,
 	}
 

proxy/tun/tun_darwin.go

@@ -11,6 +11,8 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"sync/atomic"
+	"time"
 	"unsafe"
 
 	"github.com/xtls/xray-core/common/buf"
@@ -38,21 +40,111 @@ const (
 	ND6_INFINITE_LIFETIME = 0xFFFFFFFF // netinet6/nd6.h
 )
 
-//go:linkname procyield runtime.procyield
-func procyield(cycles uint32)
-
 type DarwinTun struct {
 	tunFile *os.File
 	options *Config
 	tunFd   int
 	ownsFd  bool // true for macOS (we created the fd), false for iOS (fd from system)
 
+	// Genuinely blocks Wait() until tunFd is readable, instead of the
+	// previous procyield-only busy-spin (dispatchLoop in
+	// stack_gvisor_endpoint.go calls ReadPacket() then Wait() in a tight
+	// loop with no other throttling whenever the queue is empty -- with
+	// only procyield(1), that pins a full CPU core for as long as the
+	// tunnel is up, observed causing severe device heating/thermal
+	// shutdown). nil if kqueue setup failed, in which case Wait() falls
+	// back to a bounded time.Sleep instead. See waitKqueue's own doc
+	// comment for why this is a dedicated type rather than a bare fd.
+	waitKq *waitKqueue
+
 	routeMonitor     *os.File
 	routeMonitorOnce sync.Once
 	systemRoutes     []netip.Prefix
 	gateway          netip.Prefix
 }
 
+// waitKqueue owns a kqueue fd used by DarwinTun.Wait() to block on
+// read-readiness. Closing and waiting can race from different goroutines
+// (Close() from the caller that tears down the tunnel, Wait() from
+// dispatchLoop's own goroutine) -- reviewer feedback on XTLS/Xray-core#6580
+// found that a bare `int` fd field let Close() race Wait()'s use of the
+// same fd number, and on Darwin a closed fd number can be reused by an
+// unrelated concurrent open() before Wait() gets to call Kevent on it,
+// so Wait() could end up polling (or Close() could end up closing) a
+// completely unrelated file descriptor. This type makes closing
+// idempotent (sync.Once) and gates every Kevent call behind an atomic
+// "closed" flag checked immediately before the syscall, so Wait() never
+// issues a kevent syscall against a fd number that Close() has already
+// (or is concurrently) invalidated -- there's still a narrow window where
+// Wait() checks-then-uses the fd, but Close() only actually closes it
+// after Wait() cannot start a new syscall on it (the flag is set first,
+// synchronized with acquire/release semantics), which is sufficient since
+// Wait()'s Kevent call itself is what's being raced, not a fd read/write.
+type waitKqueue struct {
+	fd     int
+	closed atomic.Bool
+	once   sync.Once
+}
+
+// newWaitKqueue creates a kqueue registered for read-readiness on fd, for
+// Wait() to block on. Returns nil if anything fails, so callers can fall
+// back to a bounded sleep rather than error out of NewTun over what is
+// purely a CPU-efficiency concern.
+func newWaitKqueue(fd int) *waitKqueue {
+	kq, err := unix.Kqueue()
+	if err != nil {
+		return nil
+	}
+	_, err = unix.Kevent(kq, []unix.Kevent_t{{
+		Ident:  uint64(fd),
+		Filter: unix.EVFILT_READ,
+		Flags:  unix.EV_ADD | unix.EV_ENABLE,
+	}}, nil, nil)
+	if err != nil {
+		_ = unix.Close(kq)
+		return nil
+	}
+	return &waitKqueue{fd: kq}
+}
+
+// wait blocks until the registered fd is readable, timeout elapses, or a
+// benign interrupt occurs -- all three are "this kqueue is still healthy,
+// the caller should just try again" and return true; the caller
+// (DarwinTun.Wait) doesn't need to distinguish them since it always calls
+// ReadPacket() right after anyway, and that already handles "nothing was
+// actually there" via ErrQueueEmpty. Returns false only when the kqueue
+// itself is no longer usable -- already closed, or the kevent syscall
+// failed for a reason other than EINTR -- see its own call site in
+// DarwinTun.Wait for why a persistent failure must not be silently
+// retried forever (reviewer feedback, XTLS/Xray-core#6580 P2).
+func (w *waitKqueue) wait(timeout time.Duration) (ok bool) {
+	if w.closed.Load() {
+		return false
+	}
+	events := make([]unix.Kevent_t, 1)
+	ts := unix.NsecToTimespec(timeout.Nanoseconds())
+	_, err := unix.Kevent(w.fd, nil, events, &ts)
+	if err != nil {
+		return errors.Is(err, unix.EINTR)
+	}
+	return true
+}
+
+// close marks the kqueue as unusable (so any Wait() call that hasn't yet
+// entered the kevent syscall bails out instead) and closes the underlying
+// fd exactly once, regardless of how many times close is called or
+// whether it races a Wait() already inside its kevent syscall (that call
+// either completes against the still-open fd or returns an error safely
+// -- either way, no other goroutine can be handed this fd number in
+// between the atomic flag flip and the actual close, since nothing else
+// in this type ever creates a new kqueue with the same field).
+func (w *waitKqueue) close() {
+	w.once.Do(func() {
+		w.closed.Store(true)
+		_ = unix.Close(w.fd)
+	})
+}
+
 var (
 	_ Tun          = (*DarwinTun)(nil)
 	_ GVisorDevice = (*DarwinTun)(nil)
@@ -77,6 +169,7 @@ func NewTun(options *Config) (Tun, error) {
 			options: options,
 			tunFd:   fd,
 			ownsFd:  false,
+			waitKq:  newWaitKqueue(fd),
 		}, nil
 	}
 
@@ -103,6 +196,7 @@ func NewTun(options *Config) (Tun, error) {
 		options: options,
 		tunFd:   int(tunFile.Fd()),
 		ownsFd:  true,
+		waitKq:  newWaitKqueue(int(tunFile.Fd())),
 		gateway: gateway,
 	}, nil
 }
@@ -134,6 +228,9 @@ func (t *DarwinTun) Close() error {
 			_ = t.routeMonitor.Close()
 		}
 	})
+	if t.waitKq != nil {
+		t.waitKq.close()
+	}
 	routeErr := t.unsetSystemRoutes()
 	if t.ownsFd {
 		return xerrors.Combine(routeErr, t.tunFile.Close())
@@ -242,9 +339,43 @@ func (t *DarwinTun) ReadPacket() (byte, *stack.PacketBuffer, error) {
 	}), nil
 }
 
-// Wait some cpu cycles
+// Wait blocks until tunFd is readable (or a short timeout elapses), rather
+// than spinning the CPU -- see the waitKq field's own doc comment. A bounded
+// timeout (not an indefinite wait) keeps this responsive to a Close() that
+// happens to race a call already parked here.
+//
+// Reviewer feedback (XTLS/Xray-core#6580, P2): the original version
+// discarded every error from the underlying kevent syscall. dispatchLoop
+// (stack_gvisor_endpoint.go) calls ReadPacket() then Wait() in an
+// unconditional tight loop -- if kevent started failing at runtime for a
+// persistent reason (not just a benign EINTR), Wait() returning
+// immediately every time reintroduces exactly the busy-spin this whole
+// change exists to remove, just routed through a failing syscall instead
+// of procyield. waitKq.wait's own bool return distinguishes "genuinely
+// interrupted, try again" from "this kqueue is unusable now" -- Wait()
+// permanently falls back to the sleep path once that happens, rather than
+// retrying the same broken kqueue forever.
 func (t *DarwinTun) Wait() {
-	procyield(1)
+	if t.waitKq != nil && t.waitKq.wait(time.Second) {
+		return
+	}
+	if t.waitKq != nil {
+		// Persistent kevent failure (not a benign EINTR, and not just
+		// "the 1s timeout elapsed with nothing to read" -- wait() already
+		// returned true for both of those cases above). Stop trusting
+		// this kqueue for the rest of this DarwinTun's lifetime instead of
+		// re-attempting a syscall that's already shown it won't succeed.
+		t.waitKq.close()
+		t.waitKq = nil
+	}
+	// Reviewer feedback (XTLS/Xray-core#6580): procyield here is the same
+	// busy-spin this whole change exists to remove, just gated behind an
+	// edge case (kqueue setup failing, which practically never happens on
+	// real Darwin systems, or having just failed permanently above)
+	// instead of always -- a genuine bounded sleep actually yields the CPU
+	// instead of being a near-instant scheduler hint that lets the tight
+	// dispatchLoop caller spin just as hot as before.
+	time.Sleep(time.Millisecond)
 }
 
 func (t *DarwinTun) newEndpoint() (stack.LinkEndpoint, error) {

proxy/tun/tun_darwin_test.go

@@ -3,7 +3,11 @@
 package tun
 
 import (
+	"sync"
 	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
 )
 
 func TestSelectDarwinGatewayDefault(t *testing.T) {
@@ -47,3 +51,183 @@ func TestSelectDarwinGatewayRequiresUsableLocalAddress(t *testing.T) {
 		t.Fatal("expected error")
 	}
 }
+
+// newTestSocketpair returns a connected AF_UNIX/SOCK_DGRAM pair -- a real
+// fd DarwinTun.Wait's kqueue can register EVFILT_READ against, without
+// needing an actual utun interface (which requires root/network
+// entitlements this test environment doesn't have). Datagram sockets
+// (unlike pipes) support both "write makes readable" and "close makes
+// readable" the same way a tun fd's read-readiness behaves.
+func newTestSocketpair(t *testing.T) (a, b int) {
+	t.Helper()
+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0)
+	if err != nil {
+		t.Fatalf("socketpair: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = unix.Close(fds[0])
+		_ = unix.Close(fds[1])
+	})
+	return fds[0], fds[1]
+}
+
+// Reviewer feedback, XTLS/Xray-core#6580: "blocking with no data" case --
+// wait() must not return before the timeout when nothing is written.
+func TestWaitKqueueBlocksWithNoData(t *testing.T) {
+	a, _ := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+	defer kq.close()
+
+	start := time.Now()
+	ok := kq.wait(150 * time.Millisecond)
+	elapsed := time.Since(start)
+
+	if !ok {
+		t.Fatal("wait() returned false on a healthy kqueue with a plain timeout")
+	}
+	if elapsed < 100*time.Millisecond {
+		t.Fatalf("wait() returned after only %v, expected it to block close to the 150ms timeout", elapsed)
+	}
+}
+
+// Reviewer feedback: "wake up with a readable fd" case.
+func TestWaitKqueueWakesOnReadable(t *testing.T) {
+	a, b := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+	defer kq.close()
+
+	done := make(chan bool, 1)
+	go func() {
+		done <- kq.wait(5 * time.Second)
+	}()
+
+	time.Sleep(20 * time.Millisecond) // let wait() actually enter the syscall first
+	if _, err := unix.Write(b, []byte{0x1}); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	select {
+	case ok := <-done:
+		if !ok {
+			t.Fatal("wait() returned false after the fd became readable")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("wait() did not wake up within 2s of the fd becoming readable")
+	}
+}
+
+// Reviewer feedback: "timeout" case, explicitly (distinct from the
+// no-data test above, which also checks blocking duration -- this one
+// only checks the return value).
+func TestWaitKqueueTimesOut(t *testing.T) {
+	a, _ := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+	defer kq.close()
+
+	if !kq.wait(50 * time.Millisecond) {
+		t.Fatal("wait() returned false on a plain timeout with no error condition")
+	}
+}
+
+// Reviewer feedback: "Close() wakes a blocked wait" case, and the
+// no-double-close/no-fd-reuse concern (P1) -- close() while wait() is
+// parked in its syscall must not panic, must not leave wait() hung, and a
+// second close() call (from a caller that, say, calls Close() twice on
+// the same DarwinTun) must be safe.
+func TestWaitKqueueCloseDuringWaitIsSafe(t *testing.T) {
+	a, _ := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+
+	started := make(chan struct{})
+	done := make(chan bool, 1)
+	go func() {
+		close(started)
+		done <- kq.wait(5 * time.Second)
+	}()
+
+	<-started
+	time.Sleep(20 * time.Millisecond) // let wait() actually enter the syscall first
+	kq.close()
+	kq.close() // double-close must be idempotent, not panic or double-free the fd
+
+	select {
+	case <-done:
+		// Either true (the close-of-the-underlying-fd unblocked kevent, a
+		// real kqueue behavior) or false (wait() observed the closed flag
+		// first) is acceptable -- what matters is that it returned at all,
+		// promptly, without hanging or crashing.
+	case <-time.After(2 * time.Second):
+		t.Fatal("wait() did not return within 2s of close() being called")
+	}
+
+	// A wait() call *after* close() must return false immediately (the
+	// closed-flag fast path), not attempt a syscall against the
+	// already-closed (and potentially since-reused, on a real system) fd
+	// number.
+	if kq.wait(time.Second) {
+		t.Fatal("wait() returned true after close() -- should short-circuit via the closed flag")
+	}
+}
+
+// Reviewer feedback: "multiple/concurrent close guard" case -- many
+// goroutines calling close() concurrently must close the underlying fd
+// exactly once.
+func TestWaitKqueueConcurrentCloseIsSafe(t *testing.T) {
+	a, _ := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+
+	var wg sync.WaitGroup
+	for range 20 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			kq.close()
+		}()
+	}
+	wg.Wait()
+
+	if !kq.closed.Load() {
+		t.Fatal("closed flag not set after concurrent close() calls")
+	}
+}
+
+// Reviewer feedback: "kevent runtime failure without spinning" case (P2).
+// Simulates a kqueue that has gone bad (closed out from under it, as if a
+// concurrent/erroneous close happened) and confirms wait() reports it as
+// unusable (false) rather than silently returning true forever, which is
+// what DarwinTun.Wait relies on to permanently fall back to the sleep
+// path instead of re-entering a failing syscall on every dispatchLoop
+// iteration.
+func TestWaitKqueueReportsPersistentFailure(t *testing.T) {
+	a, _ := newTestSocketpair(t)
+	kq := newWaitKqueue(a)
+	if kq == nil {
+		t.Fatal("newWaitKqueue returned nil")
+	}
+	// Close the underlying kqueue fd directly (bypassing kq.close(), which
+	// would also set the closed flag) to simulate the fd going bad for a
+	// reason other than this type's own close() -- e.g. some other code
+	// path in the process closing it, or the kernel invalidating it.
+	_ = unix.Close(kq.fd)
+
+	for i := 0; i < 5; i++ {
+		if kq.wait(50 * time.Millisecond) {
+			t.Fatalf("wait() call %d returned true against a closed underlying fd -- should report failure, not spin", i)
+		}
+	}
+}

transport/internet/sockopt_openbsd.go

@@ -14,11 +14,18 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 
 func applyInboundSocketOptions(network string, fd uintptr, config *SocketConfig) error {
 	if config.ReceiveOriginalDestAddress && isUDPSocket(network) {
-		if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVDSTADDR, 1); err != nil {
+		// Only the options matching the socket's address family are accepted,
-			return errors.New("failed to set IP_RECVDSTADDR").Base(err)
+		// so one of the two pairs succeeding is enough.
+		err6 := unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_RECVPKTINFO, 1)
+		if err6 == nil {
+			err6 = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_RECVDSTPORT, 1)
 		}
-		if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVDSTPORT, 1); err != nil {
+		err4 := unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVDSTADDR, 1)
-			return errors.New("failed to set IP_RECVDSTPORT").Base(err)
+		if err4 == nil {
+			err4 = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVDSTPORT, 1)
+		}
+		if err4 != nil && err6 != nil {
+			return errors.New("failed to enable receiving the original destination").Base(err4)
 		}
 	}
 	return nil

transport/internet/splithttp/client.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/http/httptrace"
 	"sync"
+	"sync/atomic"
 
 	"github.com/apernet/quic-go/http3"
 	"github.com/xtls/xray-core/common"
@@ -32,7 +33,7 @@ type DialerClient interface {
 type DefaultDialerClient struct {
 	transportConfig *Config
 	client          *http.Client
-	closed          bool
+	closed          atomic.Bool
 	httpVersion     string
 	// pool of net.Conn, created using dialUploadConn
 	uploadRawPool  *sync.Pool
@@ -40,7 +41,7 @@ type DefaultDialerClient struct {
 }
 
 func (c *DefaultDialerClient) IsClosed() bool {
-	return c.closed
+	return c.closed.Load()
 }
 
 func (c *DefaultDialerClient) OpenStream(ctx context.Context, url string, sessionId string, body io.Reader, uploadOnly bool) (wrc io.ReadCloser, remoteAddr, localAddr net.Addr, err error) {
@@ -72,7 +73,7 @@ func (c *DefaultDialerClient) OpenStream(ctx context.Context, url string, sessio
 		resp, err := c.client.Do(req)
 		if err != nil {
 			if !uploadOnly { // stream-down is enough
-				c.closed = true
+				c.closed.Store(true)
 				errors.LogInfoInner(ctx, err, "failed to "+method+" "+url)
 			}
 			gotConn.Close()
@@ -108,7 +109,7 @@ func (c *DefaultDialerClient) PostPacket(ctx context.Context, url string, sessio
 	if c.httpVersion != "1.1" {
 		resp, err := c.client.Do(req)
 		if err != nil {
-			c.closed = true
+			c.closed.Store(true)
 			return err
 		}
 
@@ -148,7 +149,7 @@ func (c *DefaultDialerClient) PostPacket(ctx context.Context, url string, sessio
 				if h1UploadConn.UnreadedResponsesCount > 0 {
 					resp, err := http.ReadResponse(h1UploadConn.RespBufReader, req)
 					if err != nil {
-						c.closed = true
+						c.closed.Store(true)
 						return fmt.Errorf("error while reading response: %s", err.Error())
 					}
 					io.Copy(io.Discard, resp.Body)

transport/internet/splithttp/config.go

@@ -1,6 +1,7 @@
 package splithttp
 
 import (
+	"bytes"
 	"encoding/base64"
 	"fmt"
 	"io"
@@ -330,14 +331,18 @@ func (c *Config) FillStreamRequest(request *http.Request, sessionId string, seqS
 func (c *Config) FillPacketRequest(request *http.Request, sessionId string, seqStr string, payload buf.MultiBuffer) error {
 	dataPlacement := c.GetNormalizedUplinkDataPlacement()
 
+	data := make([]byte, payload.Len())
+	payload.Copy(data)
+	buf.ReleaseMulti(payload)
+
 	if dataPlacement == PlacementBody || dataPlacement == PlacementAuto {
 		request.Header = c.GetRequestHeader()
-		request.Body = io.NopCloser(&buf.MultiBufferContainer{MultiBuffer: payload})
+		request.Body = io.NopCloser(bytes.NewReader(data))
-		request.ContentLength = int64(payload.Len())
+		request.ContentLength = int64(len(data))
+		request.GetBody = func() (io.ReadCloser, error) {
+			return io.NopCloser(bytes.NewReader(data)), nil
+		}
 	} else {
-		data := make([]byte, payload.Len())
-		payload.Copy(data)
-		buf.ReleaseMulti(payload)
 		switch dataPlacement {
 		case PlacementHeader:
 			request.Header = c.GetRequestHeaderWithPayload(data)

transport/internet/splithttp/config_test.go

@@ -1,9 +1,13 @@
 package splithttp_test
 
 import (
+	"io"
+	"net/http"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/xtls/xray-core/common"
+	"github.com/xtls/xray-core/common/buf"
 	. "github.com/xtls/xray-core/transport/internet/splithttp"
 )
 
@@ -77,3 +81,35 @@ func Test_GetNormalizedPath(t *testing.T) {
 		})
 	}
 }
+
+func Test_FillPacketRequest_GetBody(t *testing.T) {
+	data := []byte("hello xray")
+	payload := buf.MergeBytes(nil, data)
+
+	req, err := http.NewRequest("POST", "https://example.com/", nil)
+	common.Must(err)
+
+	config := &Config{}
+	config.FillPacketRequest(req, "sess", "0", payload)
+
+	if req.GetBody == nil {
+		t.Fatalf("Expected GetBody to be set")
+	}
+
+	first, err := io.ReadAll(req.Body)
+	common.Must(err)
+
+	if string(data) != string(first) {
+		t.Fatalf("Body mismatch. Format %q and %q are not equal", data, first)
+	}
+
+	body2, err := req.GetBody()
+	common.Must(err)
+
+	second, err := io.ReadAll(body2)
+	common.Must(err)
+
+	if string(data) != string(second) {
+		t.Fatalf("Replayed body mismatch. Format %q and %q are not equal", data, second)
+	}
+}

transport/internet/splithttp/dialer.go

@@ -595,11 +595,12 @@ func (w uploadWriter) Write(b []byte) (int, error) {
 
 	var writed int
 	for _, buff := range buffer.MultiBuffer {
+		n := int(buff.Len())
 		err := w.WriteMultiBuffer(buf.MultiBuffer{buff})
 		if err != nil {
 			return writed, err
 		}
-		writed += int(buff.Len())
+		writed += n
 	}
 	return writed, nil
 }

transport/internet/tcp/hub.go

@@ -42,9 +42,6 @@ func ListenTCP(ctx context.Context, address net.Address, port net.Port, streamSe
 	var listener net.Listener
 	var err error
 	if port == net.Port(0) { // unix
-		if !address.Family().IsDomain() {
-			return nil, errors.New("invalid unix listen: ", address).AtError()
-		}
 		listener, err = internet.ListenSystem(ctx, &net.UnixAddr{
 			Name: address.Domain(),
 			Net:  "unix",

transport/internet/tcp_hub.go

@@ -65,6 +65,9 @@ func ListenTCP(ctx context.Context, address net.Address, port net.Port, settings
 	if address.Family().IsDomain() {
 		return nil, errors.New("domain address is not allowed for listening: ", address.Domain())
 	}
+	if port == 0 {
+		return nil, errors.New("port 0 is not allowed for listening on TCP")
+	}
 
 	protocol := settings.ProtocolName
 	listenFunc := transportListenerCache[protocol]

transport/internet/udp/hub_openbsd.go

@@ -17,23 +17,39 @@ func retrieveOriginalDestFromControlMessages(msgs []unix.SocketControlMessage) n
 	var havePort bool
 
 	for _, msg := range msgs {
-		if msg.Header.Level != unix.IPPROTO_IP {
+		switch msg.Header.Level {
-			continue
+		case unix.IPPROTO_IP:
-		}
+			switch msg.Header.Type {
-
+			case unix.IP_RECVDSTADDR:
-		switch msg.Header.Type {
+				if len(msg.Data) < 4 {
-		case unix.IP_RECVDSTADDR:
+					continue
-			if len(msg.Data) < 4 {
+				}
-				continue
+				ip = append(ip[:0], msg.Data[:4]...)
+				haveAddress = true
+			case unix.IP_RECVDSTPORT:
+				if len(msg.Data) < 2 {
+					continue
+				}
+				port = binary.BigEndian.Uint16(msg.Data[:2])
+				havePort = true
 			}
-			ip = append(ip[:0], msg.Data[:4]...)
+		case unix.IPPROTO_IPV6:
-			haveAddress = true
+			switch msg.Header.Type {
-		case unix.IP_RECVDSTPORT:
+			case unix.IPV6_PKTINFO:
-			if len(msg.Data) < 2 {
+				// struct in6_pktinfo: the destination address is followed by
-				continue
+				// the interface index.
+				if len(msg.Data) < 16 {
+					continue
+				}
+				ip = append(ip[:0], msg.Data[:16]...)
+				haveAddress = true
+			case unix.IPV6_RECVDSTPORT:
+				if len(msg.Data) < 2 {
+					continue
+				}
+				port = binary.BigEndian.Uint16(msg.Data[:2])
+				havePort = true
 			}
-			port = binary.BigEndian.Uint16(msg.Data[:2])
-			havePort = true
 		}
 	}
 

transport/internet/udp/hub_openbsd_test.go

@@ -33,6 +33,36 @@ func TestRetrieveOriginalDestFromControlMessages(t *testing.T) {
 	}
 }
 
+func TestRetrieveOriginalDestFromControlMessagesIPv6(t *testing.T) {
+	// 2001:db8::7 followed by the interface index, as in struct in6_pktinfo.
+	pktinfo := []byte{
+		0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0,
+		0, 0, 0, 0, 0, 0, 0, 0x07,
+		0x02, 0, 0, 0,
+	}
+	msgs := []unix.SocketControlMessage{
+		{
+			Header: unix.Cmsghdr{Level: unix.IPPROTO_IPV6, Type: unix.IPV6_RECVDSTPORT},
+			Data:   []byte{0x30, 0x39},
+		},
+		{
+			Header: unix.Cmsghdr{Level: unix.IPPROTO_IPV6, Type: unix.IPV6_PKTINFO},
+			Data:   pktinfo,
+		},
+	}
+
+	dest := retrieveOriginalDestFromControlMessages(msgs)
+	if !dest.IsValid() {
+		t.Fatal("destination is invalid")
+	}
+	if got, want := dest.Address.IP().String(), "2001:db8::7"; got != want {
+		t.Fatalf("address = %q, want %q", got, want)
+	}
+	if got, want := dest.Port.Value(), uint16(12345); got != want {
+		t.Fatalf("port = %d, want %d", got, want)
+	}
+}
+
 func TestRetrieveOriginalDestRequiresAddressAndPort(t *testing.T) {
 	tests := [][]unix.SocketControlMessage{
 		{
@@ -47,6 +77,18 @@ func TestRetrieveOriginalDestRequiresAddressAndPort(t *testing.T) {
 				Data:   []byte{0x30, 0x39},
 			},
 		},
+		{
+			{
+				Header: unix.Cmsghdr{Level: unix.IPPROTO_IPV6, Type: unix.IPV6_PKTINFO},
+				Data:   make([]byte, 20),
+			},
+		},
+		{
+			{
+				Header: unix.Cmsghdr{Level: unix.IPPROTO_IPV6, Type: unix.IPV6_RECVDSTPORT},
+				Data:   []byte{0x30, 0x39},
+			},
+		},
 	}
 
 	for _, msgs := range tests {