Stat: Fix GetOrRegister race

Co-Authored-By: Joshua Su <i@joshua.su>

app/dispatcher/default.go

@@ -162,7 +162,7 @@ func (d *DefaultDispatcher) getLink(ctx context.Context) (*transport.Link, *tran
 		p := d.policy.ForLevel(user.Level)
 		if p.Stats.UserUplink {
 			name := "user>>>" + user.Email + ">>>traffic>>>uplink"
-			if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
+			if c, _ := d.stats.GetOrRegisterCounter(name); c != nil {
 				inboundLink.Writer = &SizeStatWriter{
 					Counter: c,
 					Writer:  inboundLink.Writer,
@@ -171,7 +171,7 @@ func (d *DefaultDispatcher) getLink(ctx context.Context) (*transport.Link, *tran
 		}
 		if p.Stats.UserDownlink {
 			name := "user>>>" + user.Email + ">>>traffic>>>downlink"
-			if c, _ := stats.GetOrRegisterCounter(d.stats, name); c != nil {
+			if c, _ := d.stats.GetOrRegisterCounter(name); c != nil {
 				outboundLink.Writer = &SizeStatWriter{
 					Counter: c,
 					Writer:  outboundLink.Writer,
@@ -200,13 +200,13 @@ func WrapLink(ctx context.Context, policyManager policy.Manager, statsManager st
 		p := policyManager.ForLevel(user.Level)
 		if p.Stats.UserUplink {
 			name := "user>>>" + user.Email + ">>>traffic>>>uplink"
-			if c, _ := stats.GetOrRegisterCounter(statsManager, name); c != nil {
+			if c, _ := statsManager.GetOrRegisterCounter(name); c != nil {
 				link.Reader.(*buf.TimeoutWrapperReader).Counter = c
 			}
 		}
 		if p.Stats.UserDownlink {
 			name := "user>>>" + user.Email + ">>>traffic>>>downlink"
-			if c, _ := stats.GetOrRegisterCounter(statsManager, name); c != nil {
+			if c, _ := statsManager.GetOrRegisterCounter(name); c != nil {
 				link.Writer = &SizeStatWriter{
 					Counter: c,
 					Writer:  link.Writer,
@@ -223,7 +223,7 @@ func WrapLink(ctx context.Context, policyManager policy.Manager, statsManager st
 
 func trackOnlineIP(ctx context.Context, sm stats.Manager, email, ip string) {
 	name := "user>>>" + email + ">>>online"
-	if om, _ := stats.GetOrRegisterOnlineMap(sm, name); om != nil {
+	if om, _ := sm.GetOrRegisterOnlineMap(name); om != nil {
 		om.AddIP(ip)
 		context.AfterFunc(ctx, func() { om.RemoveIP(ip) })
 	}

app/proxyman/inbound/always.go

@@ -26,7 +26,7 @@ func getStatCounter(v *core.Instance, tag string) (stats.Counter, stats.Counter)
 	if len(tag) > 0 && policy.ForSystem().Stats.InboundUplink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + tag + ">>>traffic>>>uplink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			uplinkCounter = c
 		}
@@ -34,7 +34,7 @@ func getStatCounter(v *core.Instance, tag string) (stats.Counter, stats.Counter)
 	if len(tag) > 0 && policy.ForSystem().Stats.InboundDownlink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + tag + ">>>traffic>>>downlink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			downlinkCounter = c
 		}

app/proxyman/outbound/handler.go

@@ -39,7 +39,7 @@ func getStatCounter(v *core.Instance, tag string) (stats.Counter, stats.Counter)
 	if len(tag) > 0 && policy.ForSystem().Stats.OutboundUplink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "outbound>>>" + tag + ">>>traffic>>>uplink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			uplinkCounter = c
 		}
@@ -47,7 +47,7 @@ func getStatCounter(v *core.Instance, tag string) (stats.Counter, stats.Counter)
 	if len(tag) > 0 && policy.ForSystem().Stats.OutboundDownlink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "outbound>>>" + tag + ">>>traffic>>>downlink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			downlinkCounter = c
 		}

app/stats/stats.go

@@ -48,6 +48,20 @@ func (m *Manager) RegisterCounter(name string) (stats.Counter, error) {
 	return c, nil
 }
 
+// GetOrRegisterCounter implements stats.Manager.
+func (m *Manager) GetOrRegisterCounter(name string) (stats.Counter, error) {
+	m.access.Lock()
+	defer m.access.Unlock()
+
+	if c, found := m.counters[name]; found {
+		return c, nil
+	}
+	errors.LogDebug(context.Background(), "create new counter ", name)
+	c := new(Counter)
+	m.counters[name] = c
+	return c, nil
+}
+
 // UnregisterCounter implements stats.Manager.
 func (m *Manager) UnregisterCounter(name string) error {
 	m.access.Lock()
@@ -97,6 +111,20 @@ func (m *Manager) RegisterOnlineMap(name string) (stats.OnlineMap, error) {
 	return om, nil
 }
 
+// GetOrRegisterOnlineMap implements stats.Manager.
+func (m *Manager) GetOrRegisterOnlineMap(name string) (stats.OnlineMap, error) {
+	m.access.Lock()
+	defer m.access.Unlock()
+
+	if om, found := m.onlineMaps[name]; found {
+		return om, nil
+	}
+	errors.LogDebug(context.Background(), "create new OnlineMap ", name)
+	om := NewOnlineMap()
+	m.onlineMaps[name] = om
+	return om, nil
+}
+
 // UnregisterOnlineMap implements stats.Manager.
 func (m *Manager) UnregisterOnlineMap(name string) error {
 	m.access.Lock()
@@ -149,6 +177,26 @@ func (m *Manager) RegisterChannel(name string) (stats.Channel, error) {
 	return c, nil
 }
 
+// GetOrRegisterChannel implements stats.Manager.
+func (m *Manager) GetOrRegisterChannel(name string) (stats.Channel, error) {
+	m.access.Lock()
+	defer m.access.Unlock()
+
+	if c, found := m.channels[name]; found {
+		return c, nil
+	}
+	errors.LogDebug(context.Background(), "create new channel ", name)
+	c := NewChannel(&ChannelConfig{BufferSize: 64, Blocking: false})
+	if m.running {
+		// Start before publishing so no goroutine can observe an unstarted channel.
+		if err := c.Start(); err != nil {
+			return nil, err
+		}
+	}
+	m.channels[name] = c
+	return c, nil
+}
+
 // UnregisterChannel implements stats.Manager.
 func (m *Manager) UnregisterChannel(name string) error {
 	m.access.Lock()

common/net/find_process_darwin.go

@@ -0,0 +1,348 @@
+//go:build darwin
+
+package net
+
+import (
+	"bytes"
+	"net"
+	"net/netip"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/xtls/xray-core/common/errors"
+)
+
+const (
+	darwinProcPIDListFDs      = 1
+	darwinProcPIDFDSocketInfo = 3
+	darwinProcFDTypeSocket    = 2
+	darwinProcFDInfoSize      = 8
+	darwinSocketFDInfoSize    = 792
+	darwinSocketFDInfoPSIOff  = 24
+	darwinSocketInfoProtoOff  = darwinSocketFDInfoPSIOff + 156
+	darwinSocketInfoFamilyOff = darwinSocketFDInfoPSIOff + 160
+	darwinSocketInfoKindOff   = darwinSocketFDInfoPSIOff + 232
+	darwinSocketInfoInSockOff = darwinSocketFDInfoPSIOff + 240
+	darwinInSockInfoFPortOff  = darwinSocketInfoInSockOff
+	darwinInSockInfoLPortOff  = darwinSocketInfoInSockOff + 4
+	darwinInSockInfoVFlagOff  = darwinSocketInfoInSockOff + 24
+	darwinInSockInfoFAddrOff  = darwinSocketInfoInSockOff + 32
+	darwinInSockInfoLAddrOff  = darwinSocketInfoInSockOff + 48
+	darwinInSockInfoSize      = 80
+	darwinInSockInfoIPv4      = 0x1
+	darwinInSockInfoIPv6      = 0x2
+	darwinSockInfoIN          = 1
+	darwinSockInfoTCP         = 2
+)
+
+type darwinSocketMatchLevel int
+
+const (
+	darwinSocketNoMatch darwinSocketMatchLevel = iota
+	darwinSocketPortMatch
+	darwinSocketRemoteMatch
+	darwinSocketLocalMatch
+	darwinSocketExactMatch
+)
+
+func FindProcess(network, srcIP string, srcPort uint16, destIP string, destPort uint16) (PID int, Name string, AbsolutePath string, err error) {
+	isLocal, err := IsLocal(net.ParseIP(srcIP))
+	if err != nil {
+		return 0, "", "", errors.New("failed to determine if address is local: ", err)
+	}
+	if !isLocal {
+		return 0, "", "", ErrNotLocal
+	}
+	if network != "tcp" && network != "udp" {
+		panic("Unsupported network type for process lookup.")
+	}
+
+	srcAddr, err := netip.ParseAddr(srcIP)
+	if err != nil {
+		return 0, "", "", errors.New("invalid source IP address: ", srcIP)
+	}
+	srcAddr = srcAddr.Unmap()
+
+	var dstAddr netip.Addr
+	hasDstAddr := false
+	if destIP != "" && destPort != 0 {
+		dstAddr, err = netip.ParseAddr(destIP)
+		if err != nil {
+			return 0, "", "", errors.New("invalid destination IP address: ", destIP)
+		}
+		dstAddr = dstAddr.Unmap()
+		hasDstAddr = true
+	}
+
+	processes, err := unix.SysctlKinfoProcSlice("kern.proc.all")
+	if err != nil {
+		return 0, "", "", errors.New("failed to list processes").Base(err)
+	}
+
+	var bestPID int32
+	bestLevel := darwinSocketNoMatch
+	ambiguousBest := false
+
+	for _, process := range processes {
+		pid := process.Proc.P_pid
+		if pid <= 0 {
+			continue
+		}
+
+		matchLevel, err := darwinProcessSocketMatchLevel(pid, network, srcAddr, srcPort, dstAddr, destPort, hasDstAddr)
+		if err != nil || matchLevel == darwinSocketNoMatch {
+			continue
+		}
+		if matchLevel == darwinSocketExactMatch {
+			bestPID = pid
+			bestLevel = matchLevel
+			ambiguousBest = false
+			break
+		}
+		if matchLevel > bestLevel {
+			bestPID = pid
+			bestLevel = matchLevel
+			ambiguousBest = false
+			continue
+		}
+		if matchLevel == bestLevel {
+			ambiguousBest = true
+		}
+	}
+
+	if bestLevel == darwinSocketNoMatch {
+		return 0, "", "", errors.New("process not found for ", network, " connection from ", srcIP, ":", srcPort, " to ", destIP, ":", destPort)
+	}
+	if ambiguousBest {
+		return 0, "", "", errors.New("ambiguous process match for ", network, " connection from ", srcIP, ":", srcPort, " to ", destIP, ":", destPort)
+	}
+
+	absPath, err := darwinProcessPath(bestPID)
+	if err != nil {
+		return 0, "", "", errors.New("could not get process path for PID ", bestPID, ": ", err)
+	}
+
+	absPath = filepath.ToSlash(absPath)
+	return int(bestPID), filepath.Base(absPath), absPath, nil
+}
+
+func darwinProcessSocketMatchLevel(pid int32, network string, srcAddr netip.Addr, srcPort uint16, dstAddr netip.Addr, dstPort uint16, hasDstAddr bool) (darwinSocketMatchLevel, error) {
+	fds, err := darwinProcessFDs(pid)
+	if err != nil {
+		return darwinSocketNoMatch, err
+	}
+
+	bestLevel := darwinSocketNoMatch
+	info := make([]byte, darwinSocketFDInfoSize)
+	for fd := 0; fd+darwinProcFDInfoSize <= len(fds); fd += darwinProcFDInfoSize {
+		fdNumber := int32(darwinReadNativeUint32(fds[fd : fd+4]))
+		fdType := darwinReadNativeUint32(fds[fd+4 : fd+8])
+		if fdType != darwinProcFDTypeSocket {
+			continue
+		}
+
+		n, err := darwinProcPIDFDInfo(pid, fdNumber, darwinProcPIDFDSocketInfo, info)
+		if err != nil || n < darwinSocketInfoInSockOff+darwinInSockInfoSize {
+			continue
+		}
+		level := darwinSocketInfoMatchLevel(info[:n], network, srcAddr, srcPort, dstAddr, dstPort, hasDstAddr)
+		if level == darwinSocketExactMatch {
+			return level, nil
+		}
+		if level > bestLevel {
+			bestLevel = level
+		}
+	}
+
+	return bestLevel, nil
+}
+
+func darwinProcessFDs(pid int32) ([]byte, error) {
+	n, err := darwinProcPIDInfo(pid, darwinProcPIDListFDs, 0, nil)
+	if err != nil {
+		return nil, err
+	}
+	if n <= 0 {
+		return nil, nil
+	}
+
+	buf := make([]byte, n)
+	n, err = darwinProcPIDInfo(pid, darwinProcPIDListFDs, 0, buf)
+	if err != nil {
+		return nil, err
+	}
+	return buf[:n], nil
+}
+
+func darwinSocketInfoMatchLevel(info []byte, network string, srcAddr netip.Addr, srcPort uint16, dstAddr netip.Addr, dstPort uint16, hasDstAddr bool) darwinSocketMatchLevel {
+	protocol := int(darwinReadNativeUint32(info[darwinSocketInfoProtoOff : darwinSocketInfoProtoOff+4]))
+	family := int(darwinReadNativeUint32(info[darwinSocketInfoFamilyOff : darwinSocketInfoFamilyOff+4]))
+	kind := int(darwinReadNativeUint32(info[darwinSocketInfoKindOff : darwinSocketInfoKindOff+4]))
+
+	switch network {
+	case "tcp":
+		if protocol != unix.IPPROTO_TCP || kind != darwinSockInfoTCP {
+			return darwinSocketNoMatch
+		}
+	case "udp":
+		if protocol != unix.IPPROTO_UDP || kind != darwinSockInfoIN {
+			return darwinSocketNoMatch
+		}
+	default:
+		return darwinSocketNoMatch
+	}
+
+	vflag := info[darwinInSockInfoVFlagOff]
+	if srcAddr.Is4() {
+		if family != unix.AF_INET || vflag&darwinInSockInfoIPv4 == 0 {
+			return darwinSocketNoMatch
+		}
+	} else {
+		if family != unix.AF_INET6 || vflag&darwinInSockInfoIPv6 == 0 {
+			return darwinSocketNoMatch
+		}
+	}
+
+	localPort := int32(darwinReadNativeUint32(info[darwinInSockInfoLPortOff : darwinInSockInfoLPortOff+4]))
+	if !darwinPortMatches(localPort, srcPort) {
+		return darwinSocketNoMatch
+	}
+	localAddrMatches := darwinAddrMatchesOrUnspecified(info[darwinInSockInfoLAddrOff:darwinInSockInfoLAddrOff+16], srcAddr)
+
+	foreignAddrRaw := info[darwinInSockInfoFAddrOff : darwinInSockInfoFAddrOff+16]
+	foreignPort := int32(darwinReadNativeUint32(info[darwinInSockInfoFPortOff : darwinInSockInfoFPortOff+4]))
+
+	if !hasDstAddr {
+		if localAddrMatches {
+			return darwinSocketExactMatch
+		}
+		return darwinSocketNoMatch
+	}
+
+	remoteMatches := darwinPortMatches(foreignPort, dstPort) && darwinAddrMatches(foreignAddrRaw, dstAddr)
+	if network == "udp" && darwinEndpointIsZero(foreignAddrRaw, foreignPort) && localAddrMatches {
+		return darwinSocketExactMatch
+	}
+	switch {
+	case localAddrMatches && remoteMatches:
+		return darwinSocketExactMatch
+	case localAddrMatches:
+		return darwinSocketLocalMatch
+	case remoteMatches:
+		return darwinSocketRemoteMatch
+	default:
+		return darwinSocketPortMatch
+	}
+}
+
+func darwinPortMatches(value int32, port uint16) bool {
+	raw := uint16(value)
+	return raw == port || darwinNtohs(raw) == port
+}
+
+func darwinNtohs(value uint16) uint16 {
+	return value<<8 | value>>8
+}
+
+func darwinAddrMatches(raw []byte, addr netip.Addr) bool {
+	if addr.Is4() {
+		ip := addr.As4()
+		return bytes.Equal(raw[12:16], ip[:])
+	}
+	ip := addr.As16()
+	return bytes.Equal(raw, ip[:])
+}
+
+func darwinAddrMatchesOrUnspecified(raw []byte, addr netip.Addr) bool {
+	if darwinAddrMatches(raw, addr) {
+		return true
+	}
+	if addr.Is4() {
+		return darwinBytesAreZero(raw[12:16])
+	}
+	return darwinBytesAreZero(raw)
+}
+
+func darwinEndpointIsZero(rawAddr []byte, port int32) bool {
+	return uint32(port) == 0 && darwinBytesAreZero(rawAddr)
+}
+
+func darwinBytesAreZero(raw []byte) bool {
+	for _, value := range raw {
+		if value != 0 {
+			return false
+		}
+	}
+	return true
+}
+
+func darwinReadNativeUint32(b []byte) uint32 {
+	return *(*uint32)(unsafe.Pointer(&b[0]))
+}
+
+func darwinProcessPath(pid int32) (string, error) {
+	buf := make([]byte, unix.PathMax)
+	n, err := darwinProcPIDPath(pid, buf)
+	if err != nil {
+		return "", err
+	}
+	if n <= 0 {
+		return "", errors.New("empty process path")
+	}
+	return strings.TrimRight(string(buf[:n]), "\x00"), nil
+}
+
+func darwinProcPIDInfo(pid int32, flavor int, arg uint64, buf []byte) (int, error) {
+	var ptr unsafe.Pointer
+	if len(buf) > 0 {
+		ptr = unsafe.Pointer(&buf[0])
+	}
+
+	r0, _, errno := syscall_syscall6(libc_proc_pidinfo_trampoline_addr, uintptr(pid), uintptr(flavor), uintptr(arg), uintptr(ptr), uintptr(len(buf)), 0)
+	if errno != 0 {
+		return 0, errno
+	}
+	return int(r0), nil
+}
+
+func darwinProcPIDFDInfo(pid int32, fd int32, flavor int, buf []byte) (int, error) {
+	var ptr unsafe.Pointer
+	if len(buf) > 0 {
+		ptr = unsafe.Pointer(&buf[0])
+	}
+
+	r0, _, errno := syscall_syscall6(libc_proc_pidfdinfo_trampoline_addr, uintptr(pid), uintptr(fd), uintptr(flavor), uintptr(ptr), uintptr(len(buf)), 0)
+	if errno != 0 {
+		return 0, errno
+	}
+	return int(r0), nil
+}
+
+func darwinProcPIDPath(pid int32, buf []byte) (int, error) {
+	r0, _, errno := syscall_syscall6(libc_proc_pidpath_trampoline_addr, uintptr(pid), uintptr(unsafe.Pointer(&buf[0])), uintptr(len(buf)), 0, 0, 0)
+	if errno != 0 {
+		return 0, errno
+	}
+	return int(r0), nil
+}
+
+var libc_proc_pidinfo_trampoline_addr uintptr
+
+//go:cgo_import_dynamic libc_proc_pidinfo proc_pidinfo "/usr/lib/libproc.dylib"
+
+var libc_proc_pidfdinfo_trampoline_addr uintptr
+
+//go:cgo_import_dynamic libc_proc_pidfdinfo proc_pidfdinfo "/usr/lib/libproc.dylib"
+
+var libc_proc_pidpath_trampoline_addr uintptr
+
+//go:cgo_import_dynamic libc_proc_pidpath proc_pidpath "/usr/lib/libproc.dylib"
+
+// Implemented in the runtime package (runtime/sys_darwin.go).
+func syscall_syscall6(fn, a1, a2, a3, a4, a5, a6 uintptr) (r1, r2 uintptr, err syscall.Errno)
+
+//go:linkname syscall_syscall6 syscall.syscall6

common/net/find_process_darwin.s

@@ -0,0 +1,18 @@
+//go:build darwin
+
+#include "textflag.h"
+
+TEXT libc_proc_pidinfo_trampoline<>(SB),NOSPLIT,$0-0
+	JMP	libc_proc_pidinfo(SB)
+GLOBL	·libc_proc_pidinfo_trampoline_addr(SB), RODATA, $8
+DATA	·libc_proc_pidinfo_trampoline_addr(SB)/8, $libc_proc_pidinfo_trampoline<>(SB)
+
+TEXT libc_proc_pidfdinfo_trampoline<>(SB),NOSPLIT,$0-0
+	JMP	libc_proc_pidfdinfo(SB)
+GLOBL	·libc_proc_pidfdinfo_trampoline_addr(SB), RODATA, $8
+DATA	·libc_proc_pidfdinfo_trampoline_addr(SB)/8, $libc_proc_pidfdinfo_trampoline<>(SB)
+
+TEXT libc_proc_pidpath_trampoline<>(SB),NOSPLIT,$0-0
+	JMP	libc_proc_pidpath(SB)
+GLOBL	·libc_proc_pidpath_trampoline_addr(SB), RODATA, $8
+DATA	·libc_proc_pidpath_trampoline_addr(SB)/8, $libc_proc_pidpath_trampoline<>(SB)

common/net/find_process_darwin_test.go

@@ -0,0 +1,293 @@
+//go:build darwin
+
+package net
+
+import (
+	stdnet "net"
+	"net/netip"
+	"os"
+	"testing"
+	"time"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+func TestFindProcessDarwinTCP(t *testing.T) {
+	listener, err := stdnet.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	accepted := make(chan stdnet.Conn, 1)
+	go func() {
+		conn, err := listener.Accept()
+		if err == nil {
+			accepted <- conn
+			return
+		}
+		close(accepted)
+	}()
+
+	conn, err := stdnet.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	serverConn := <-accepted
+	if serverConn == nil {
+		t.Fatal("server did not accept tcp connection")
+	}
+	defer serverConn.Close()
+
+	local := conn.LocalAddr().(*stdnet.TCPAddr)
+	remote := conn.RemoteAddr().(*stdnet.TCPAddr)
+
+	pid, name, path, err := FindProcess("tcp", local.IP.String(), uint16(local.Port), remote.IP.String(), uint16(remote.Port))
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertCurrentProcess(t, pid, name, path)
+}
+
+func TestFindProcessDarwinUDP(t *testing.T) {
+	conn, err := stdnet.ListenUDP("udp", &stdnet.UDPAddr{IP: stdnet.ParseIP("127.0.0.1")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	local := conn.LocalAddr().(*stdnet.UDPAddr)
+
+	pid, name, path, err := FindProcess("udp", local.IP.String(), uint16(local.Port), "", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertCurrentProcess(t, pid, name, path)
+}
+
+func TestFindProcessDarwinNonLocal(t *testing.T) {
+	_, _, _, err := FindProcess("tcp", "203.0.113.1", 80, "", 0)
+	if err != ErrNotLocal {
+		t.Fatalf("expected ErrNotLocal, got %v", err)
+	}
+}
+
+func TestFindProcessDarwinUnsupportedNetwork(t *testing.T) {
+	defer func() {
+		if recover() == nil {
+			t.Fatal("expected panic")
+		}
+	}()
+
+	_, _, _, _ = FindProcess("icmp", "127.0.0.1", 0, "", 0)
+}
+
+func assertCurrentProcess(t *testing.T, pid int, name string, path string) {
+	t.Helper()
+
+	if pid != os.Getpid() {
+		t.Fatalf("expected pid %d, got %d (%s, %s)", os.Getpid(), pid, name, path)
+	}
+
+	executable, err := os.Executable()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if path == "" || name == "" {
+		t.Fatalf("expected process path and name, got name=%q path=%q", name, path)
+	}
+	if sameFile(executable, path) {
+		return
+	}
+	t.Fatalf("expected executable %q, got %q", executable, path)
+}
+
+func sameFile(left string, right string) bool {
+	leftInfo, leftErr := os.Stat(left)
+	rightInfo, rightErr := os.Stat(right)
+	if leftErr != nil || rightErr != nil {
+		return false
+	}
+	return os.SameFile(leftInfo, rightInfo)
+}
+
+func TestFindProcessDarwinTCPWithoutDestination(t *testing.T) {
+	listener, err := stdnet.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	accepted := make(chan stdnet.Conn, 1)
+	go func() {
+		conn, err := listener.Accept()
+		if err == nil {
+			accepted <- conn
+			return
+		}
+		close(accepted)
+	}()
+
+	conn, err := stdnet.DialTimeout("tcp", listener.Addr().String(), time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	serverConn := <-accepted
+	if serverConn == nil {
+		t.Fatal("server did not accept tcp connection")
+	}
+	defer serverConn.Close()
+
+	local := conn.LocalAddr().(*stdnet.TCPAddr)
+	pid, name, path, err := FindProcess("tcp", local.IP.String(), uint16(local.Port), "", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertCurrentProcess(t, pid, name, path)
+}
+
+func TestFindProcessDarwinTCPWithDifferentDestination(t *testing.T) {
+	listener, err := stdnet.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+
+	accepted := make(chan stdnet.Conn, 1)
+	go func() {
+		conn, err := listener.Accept()
+		if err == nil {
+			accepted <- conn
+			return
+		}
+		close(accepted)
+	}()
+
+	conn, err := stdnet.DialTimeout("tcp", listener.Addr().String(), time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	serverConn := <-accepted
+	if serverConn == nil {
+		t.Fatal("server did not accept tcp connection")
+	}
+	defer serverConn.Close()
+
+	local := conn.LocalAddr().(*stdnet.TCPAddr)
+	pid, name, path, err := FindProcess("tcp", local.IP.String(), uint16(local.Port), "203.0.113.10", 443)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertCurrentProcess(t, pid, name, path)
+}
+
+func TestDarwinSocketInfoMatchLevelFallbacks(t *testing.T) {
+	src := netip.MustParseAddr("198.18.0.2")
+	dst := netip.MustParseAddr("203.0.113.10")
+	otherLocal := netip.MustParseAddr("192.168.1.10")
+	otherRemote := netip.MustParseAddr("198.51.100.10")
+	unspecifiedLocal := netip.MustParseAddr("0.0.0.0")
+
+	tests := []struct {
+		name      string
+		local     netip.Addr
+		remote    netip.Addr
+		hasDst    bool
+		wantLevel darwinSocketMatchLevel
+	}{
+		{
+			name:      "exact",
+			local:     src,
+			remote:    dst,
+			hasDst:    true,
+			wantLevel: darwinSocketExactMatch,
+		},
+		{
+			name:      "unspecified local with matching remote",
+			local:     unspecifiedLocal,
+			remote:    dst,
+			hasDst:    true,
+			wantLevel: darwinSocketExactMatch,
+		},
+		{
+			name:      "unspecified local without destination",
+			local:     unspecifiedLocal,
+			remote:    otherRemote,
+			hasDst:    false,
+			wantLevel: darwinSocketExactMatch,
+		},
+		{
+			name:      "local match with different remote",
+			local:     src,
+			remote:    otherRemote,
+			hasDst:    true,
+			wantLevel: darwinSocketLocalMatch,
+		},
+		{
+			name:      "remote match with different local",
+			local:     otherLocal,
+			remote:    dst,
+			hasDst:    true,
+			wantLevel: darwinSocketRemoteMatch,
+		},
+		{
+			name:      "port only with destination",
+			local:     otherLocal,
+			remote:    otherRemote,
+			hasDst:    true,
+			wantLevel: darwinSocketPortMatch,
+		},
+		{
+			name:      "different local without destination",
+			local:     otherLocal,
+			remote:    otherRemote,
+			hasDst:    false,
+			wantLevel: darwinSocketNoMatch,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			info := newDarwinSocketInfo("tcp", test.local, 12345, test.remote, 443)
+			level := darwinSocketInfoMatchLevel(info, "tcp", src, 12345, dst, 443, test.hasDst)
+			if level != test.wantLevel {
+				t.Fatalf("unexpected match level: got %d, want %d", level, test.wantLevel)
+			}
+		})
+	}
+}
+
+func newDarwinSocketInfo(network string, local netip.Addr, localPort uint16, remote netip.Addr, remotePort uint16) []byte {
+	info := make([]byte, darwinSocketFDInfoSize)
+	switch network {
+	case "tcp":
+		writeDarwinNativeUint32(info, darwinSocketInfoProtoOff, uint32(unix.IPPROTO_TCP))
+		writeDarwinNativeUint32(info, darwinSocketInfoKindOff, uint32(darwinSockInfoTCP))
+	case "udp":
+		writeDarwinNativeUint32(info, darwinSocketInfoProtoOff, uint32(unix.IPPROTO_UDP))
+		writeDarwinNativeUint32(info, darwinSocketInfoKindOff, uint32(darwinSockInfoIN))
+	}
+	writeDarwinNativeUint32(info, darwinSocketInfoFamilyOff, uint32(unix.AF_INET))
+	info[darwinInSockInfoVFlagOff] = darwinInSockInfoIPv4
+	writeDarwinNativeUint32(info, darwinInSockInfoLPortOff, uint32(localPort))
+	writeDarwinNativeUint32(info, darwinInSockInfoFPortOff, uint32(remotePort))
+	copyDarwinIPv4(info[darwinInSockInfoLAddrOff:darwinInSockInfoLAddrOff+16], local)
+	copyDarwinIPv4(info[darwinInSockInfoFAddrOff:darwinInSockInfoFAddrOff+16], remote)
+	return info
+}
+
+func writeDarwinNativeUint32(b []byte, offset int, value uint32) {
+	*(*uint32)(unsafe.Pointer(&b[offset])) = value
+}
+
+func copyDarwinIPv4(dst []byte, addr netip.Addr) {
+	ip := addr.As4()
+	copy(dst[12:16], ip[:])
+}

common/net/find_process_others.go

@@ -1,4 +1,4 @@
-//go:build !windows && !linux && !android
+//go:build !windows && !linux && !android && !darwin
 
 package net
 

features/stats/stats.go

@@ -81,6 +81,8 @@ type Manager interface {
 
 	// RegisterCounter registers a new counter to the manager. The identifier string must not be empty, and unique among other counters.
 	RegisterCounter(string) (Counter, error)
+	// GetOrRegisterCounter returns the counter by its identifier, atomically creating and registering it if absent.
+	GetOrRegisterCounter(string) (Counter, error)
 	// UnregisterCounter unregisters a counter from the manager by its identifier.
 	UnregisterCounter(string) error
 	// GetCounter returns a counter by its identifier.
@@ -91,6 +93,8 @@ type Manager interface {
 
 	// RegisterOnlineMap registers a new OnlineMap to the manager. The identifier string must not be empty, and unique among other OnlineMaps.
 	RegisterOnlineMap(string) (OnlineMap, error)
+	// GetOrRegisterOnlineMap returns the OnlineMap by its identifier, atomically creating and registering it if absent.
+	GetOrRegisterOnlineMap(string) (OnlineMap, error)
 	// UnregisterOnlineMap unregisters an OnlineMap from the manager by its identifier.
 	UnregisterOnlineMap(string) error
 	// GetOnlineMap returns an OnlineMap by its identifier.
@@ -101,6 +105,8 @@ type Manager interface {
 
 	// RegisterChannel registers a new channel to the manager. The identifier string must not be empty, and unique among other channels.
 	RegisterChannel(string) (Channel, error)
+	// GetOrRegisterChannel returns the channel by its identifier, atomically creating and registering it if absent.
+	GetOrRegisterChannel(string) (Channel, error)
 	// UnregisterChannel unregisters a channel from the manager by its identifier.
 	UnregisterChannel(string) error
 	// GetChannel returns a channel by its identifier.
@@ -110,36 +116,6 @@ type Manager interface {
 	GetAllOnlineUsers() []string
 }
 
-// GetOrRegisterCounter tries to get the StatCounter first. If not exist, it then tries to create a new counter.
-func GetOrRegisterCounter(m Manager, name string) (Counter, error) {
-	counter := m.GetCounter(name)
-	if counter != nil {
-		return counter, nil
-	}
-
-	return m.RegisterCounter(name)
-}
-
-// GetOrRegisterOnlineMap tries to get the OnlineMap first. If not exist, it then tries to create a new OnlineMap.
-func GetOrRegisterOnlineMap(m Manager, name string) (OnlineMap, error) {
-	onlineMap := m.GetOnlineMap(name)
-	if onlineMap != nil {
-		return onlineMap, nil
-	}
-
-	return m.RegisterOnlineMap(name)
-}
-
-// GetOrRegisterChannel tries to get the StatChannel first. If not exist, it then tries to create a new channel.
-func GetOrRegisterChannel(m Manager, name string) (Channel, error) {
-	channel := m.GetChannel(name)
-	if channel != nil {
-		return channel, nil
-	}
-
-	return m.RegisterChannel(name)
-}
-
 // ManagerType returns the type of Manager interface. Can be used to implement common.HasType.
 //
 // xray:api:stable
@@ -160,6 +136,11 @@ func (NoopManager) RegisterCounter(string) (Counter, error) {
 	return nil, errors.New("not implemented")
 }
 
+// GetOrRegisterCounter implements Manager.
+func (NoopManager) GetOrRegisterCounter(string) (Counter, error) {
+	return nil, errors.New("not implemented")
+}
+
 // UnregisterCounter implements Manager.
 func (NoopManager) UnregisterCounter(string) error {
 	return nil
@@ -178,6 +159,11 @@ func (NoopManager) RegisterOnlineMap(string) (OnlineMap, error) {
 	return nil, errors.New("not implemented")
 }
 
+// GetOrRegisterOnlineMap implements Manager.
+func (NoopManager) GetOrRegisterOnlineMap(string) (OnlineMap, error) {
+	return nil, errors.New("not implemented")
+}
+
 // UnregisterOnlineMap implements Manager.
 func (NoopManager) UnregisterOnlineMap(string) error {
 	return nil
@@ -196,6 +182,11 @@ func (NoopManager) RegisterChannel(string) (Channel, error) {
 	return nil, errors.New("not implemented")
 }
 
+// GetOrRegisterChannel implements Manager.
+func (NoopManager) GetOrRegisterChannel(string) (Channel, error) {
+	return nil, errors.New("not implemented")
+}
+
 // UnregisterChannel implements Manager.
 func (NoopManager) UnregisterChannel(string) error {
 	return nil

proxy/tun/handler.go

@@ -65,7 +65,7 @@ func (t *Handler) Init(ctx context.Context, pm policy.Manager, dispatcher routin
 	if len(t.tag) > 0 && pm.ForSystem().Stats.InboundUplink {
 		statsManager := core.MustFromContext(ctx).GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + t.tag + ">>>traffic>>>uplink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			t.uplinkCounter = c
 		}
@@ -73,7 +73,7 @@ func (t *Handler) Init(ctx context.Context, pm policy.Manager, dispatcher routin
 	if len(t.tag) > 0 && pm.ForSystem().Stats.InboundDownlink {
 		statsManager := core.MustFromContext(ctx).GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + t.tag + ">>>traffic>>>downlink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			t.downlinkCounter = c
 		}

proxy/wireguard/client.go

@@ -57,7 +57,7 @@ func NewClient(ctx context.Context, conf *DeviceConfig) (*Handler, error) {
 	if len(tag) > 0 && p.ForSystem().Stats.OutboundUplink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "outbound>>>" + tag + ">>>traffic>>>uplink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			uplinkCounter = c
 		}
@@ -65,7 +65,7 @@ func NewClient(ctx context.Context, conf *DeviceConfig) (*Handler, error) {
 	if len(tag) > 0 && p.ForSystem().Stats.OutboundDownlink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "outbound>>>" + tag + ">>>traffic>>>downlink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			downlinkCounter = c
 		}

proxy/wireguard/server.go

@@ -66,7 +66,7 @@ func NewServer(ctx context.Context, conf *DeviceConfig) (*Server, error) {
 	if len(tag) > 0 && p.ForSystem().Stats.InboundUplink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + tag + ">>>traffic>>>uplink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			uplinkCounter = c
 		}
@@ -74,7 +74,7 @@ func NewServer(ctx context.Context, conf *DeviceConfig) (*Server, error) {
 	if len(tag) > 0 && p.ForSystem().Stats.InboundDownlink {
 		statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager)
 		name := "inbound>>>" + tag + ">>>traffic>>>downlink"
-		c, _ := stats.GetOrRegisterCounter(statsManager, name)
+		c, _ := statsManager.GetOrRegisterCounter(name)
 		if c != nil {
 			downlinkCounter = c
 		}

transport/internet/splithttp/config.go

@@ -24,8 +24,11 @@ func (c *Config) GetNormalizedPath() string {
 		path = "/" + path
 	}
 
-	if path[len(path)-1] != '/' {
-		path = path + "/"
+	if c.GetNormalizedSessionPlacement() == PlacementPath ||
+		c.GetNormalizedSeqPlacement() == PlacementPath {
+		if path[len(path)-1] != '/' {
+			path = path + "/"
+		}
 	}
 
 	return path

transport/internet/splithttp/config_test.go

@@ -3,16 +3,77 @@ package splithttp_test
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	. "github.com/xtls/xray-core/transport/internet/splithttp"
 )
 
 func Test_GetNormalizedPath(t *testing.T) {
-	c := Config{
-		Path: "/?world",
+	tests := []struct {
+		TestName           string
+		Path               string
+		SessionIDPlacement string
+		SeqPlacement       string
+		Expected           string
+	}{
+		{
+			TestName: "default placement keeps trailing slash",
+			Path:     "/sh",
+			Expected: "/sh/",
+		},
+		{
+			TestName: "query string is stripped",
+			Path:     "/?world",
+			Expected: "/",
+		},
+		{
+			TestName:           "both off path drops trailing slash",
+			Path:               "/stream",
+			SessionIDPlacement: "query",
+			SeqPlacement:       "query",
+			Expected:           "/stream",
+		},
+		{
+			TestName:           "both off path keeps file-like path",
+			Path:               "/stream/filename.extension",
+			SessionIDPlacement: "query",
+			SeqPlacement:       "header",
+			Expected:           "/stream/filename.extension",
+		},
+		{
+			TestName:           "seq in path keeps trailing slash",
+			Path:               "/stream",
+			SessionIDPlacement: "query",
+			Expected:           "/stream/",
+		},
+		{
+			TestName:     "session in path keeps trailing slash",
+			Path:         "/stream",
+			SeqPlacement: "cookie",
+			Expected:     "/stream/",
+		},
+		{
+			TestName:           "existing trailing slash preserved",
+			Path:               "/stream/",
+			SessionIDPlacement: "query",
+			SeqPlacement:       "query",
+			Expected:           "/stream/",
+		},
+		{
+			TestName:           "root unchanged",
+			Path:               "/",
+			SessionIDPlacement: "query",
+			SeqPlacement:       "query",
+			Expected:           "/",
+		},
 	}
-
-	path := c.GetNormalizedPath()
-	if path != "/" {
-		t.Error("Unexpected: ", path)
+	for _, test := range tests {
+		t.Run(test.TestName, func(t *testing.T) {
+			c := Config{
+				Path:               test.Path,
+				SessionIDPlacement: test.SessionIDPlacement,
+				SeqPlacement:       test.SeqPlacement,
+			}
+			assert.Equal(t, test.Expected, c.GetNormalizedPath())
+		})
 	}
 }

transport/internet/tls/config.go

@@ -450,7 +450,7 @@ func (c *Config) GetTLSConfig(opts ...Option) *tls.Config {
 		for _, s := range tls.CipherSuites() {
 			id[s.Name] = s.ID
 		}
-		for _, n := range strings.Split(c.CipherSuites, ":") {
+		for n := range strings.SplitSeq(c.CipherSuites, ":") {
 			if id[n] != 0 {
 				config.CipherSuites = append(config.CipherSuites, id[n])
 			}

transport/internet/tls/ech.go

@@ -55,19 +55,16 @@ func ApplyECH(c *Config, config *tls.Config) error {
 			}
 			config.EncryptedClientHelloConfigList = ECHConfig
 		}()
-		// direct base64 config
+		// query config from dns
 		if strings.Contains(c.EchConfigList, "://") {
-			// query config from dns
-			parts := strings.Split(c.EchConfigList, "+")
+			// parse ECH DNS server in format of "example.com+https://1.1.1.1/dns-query"
+			parts := strings.SplitN(c.EchConfigList, "+", 2)
 			if len(parts) == 2 {
-				// parse ECH DNS server in format of "example.com+https://1.1.1.1/dns-query"
 				nameToQuery = parts[0]
 				DNSServer = parts[1]
-			} else if len(parts) == 1 {
+			} else {
 				// normal format
 				DNSServer = parts[0]
-			} else {
-				return errors.New("Invalid ECH DNS server format: ", c.EchConfigList)
 			}
 			if nameToQuery == "" {
 				return errors.New("Using DNS for ECH Config needs serverName or use Server format example.com+https://1.1.1.1/dns-query")
@@ -77,6 +74,7 @@ func ApplyECH(c *Config, config *tls.Config) error {
 				return errors.New("Failed to query ECH DNS record for domain: ", nameToQuery, " at server: ", DNSServer).Base(err)
 			}
 		} else {
+			// direct base64 config
 			ECHConfig, err = base64.StdEncoding.DecodeString(c.EchConfigList)
 			if err != nil {
 				return errors.New("Failed to unmarshal ECHConfigList: ", err)
@@ -157,7 +155,7 @@ func QueryRecord(domain string, server string, sockopt *internet.SocketConfig) (
 	// If expire is zero value, it means we are in initial state, wait for the query to finish
 	// otherwise return old value immediately and update in a goroutine
 	// but if the cache is too old, wait for update
-	if configRecord.expire == (time.Time{}) || configRecord.expire.Add(time.Hour*4).Before(time.Now()) {
+	if configRecord.expire.IsZero() || configRecord.expire.Add(time.Hour*4).Before(time.Now()) {
 		return echConfigCache.Update(domain, server, false, sockopt)
 	} else {
 		// If someone already acquired the lock, it means it is updating, do not start another update goroutine
@@ -256,14 +254,17 @@ func dnsQuery(server string, domain string, sockopt *internet.SocketConfig) ([]b
 		}
 		dnsResolve = respBody
 	} else if strings.HasPrefix(server, "udp://") { // for classic udp dns server
-		udpServerAddr := server[len("udp://"):]
-		// default port 53 if not specified
-		if !strings.Contains(udpServerAddr, ":") {
-			udpServerAddr = udpServerAddr + ":53"
-		}
-		dest, err := net.ParseDestination("udp" + ":" + udpServerAddr)
+		udpServerURL, err := url.Parse(server)
 		if err != nil {
-			return nil, 0, errors.New("failed to parse udp dns server ", udpServerAddr, " for ECH: ", err)
+			return nil, 0, errors.New("failed to parse udp dns server ", server, " for ECH: ", err)
+		}
+		// default port 53 if not specified
+		if udpServerURL.Port() == "" {
+			udpServerURL.Host = udpServerURL.Host + ":53"
+		}
+		dest, err := net.ParseDestination("udp" + ":" + udpServerURL.Host)
+		if err != nil {
+			return nil, 0, errors.New("failed to parse udp dns server ", udpServerURL.Host, " for ECH: ", err)
 		}
 		dnsTimeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()

transport/internet/tls/ech_test.go

@@ -19,8 +19,7 @@ func TestECHDial(t *testing.T) {
 	// test concurrent Dial(to test cache problem)
 	wg := sync.WaitGroup{}
 	for range 10 {
-		wg.Add(1)
-		go func() {
+		wg.Go(func() {
 			TLSConfig := config.GetTLSConfig()
 			TLSConfig.NextProtos = []string{"http/1.1"}
 			client := &http.Client{
@@ -36,8 +35,7 @@ func TestECHDial(t *testing.T) {
 			if !strings.Contains(string(body), "sni=encrypted") {
 				t.Error("ECH Dial success but SNI is not encrypted")
 			}
-			wg.Done()
-		}()
+		})
 	}
 	wg.Wait()
 	// check cache