Compare commits

..
Author SHA1 Message Date
Fangliding 73baf47358 prevent close of closed 2025-08-25 13:55:17 +08:00
Fangliding ecc2f73108 refactor dns outbound conn 2025-08-25 13:28:25 +08:00
Fangliding f45ca197a2 Fix dead lock 2025-08-25 13:16:51 +08:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
33272a0499 Bump google.golang.org/protobuf from 1.36.7 to 1.36.8 (#5042)
Bumps google.golang.org/protobuf from 1.36.7 to 1.36.8.

---
updated-dependencies:
- dependency-name: google.golang.org/protobuf
  dependency-version: 1.36.8
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2025-08-21 10:36:38 +00:00
4 changed files with 89 additions and 35 deletions
+1 -1
View File
@@ -27,7 +27,7 @@ require (
golang.org/x/sys v0.35.0 golang.org/x/sys v0.35.0
golang.zx2c4.com/wireguard v0.0.0-20231211153847-12269c276173 golang.zx2c4.com/wireguard v0.0.0-20231211153847-12269c276173
google.golang.org/grpc v1.75.0 google.golang.org/grpc v1.75.0
google.golang.org/protobuf v1.36.7 google.golang.org/protobuf v1.36.8
gvisor.dev/gvisor v0.0.0-20250428193742-2d800c3129d5 gvisor.dev/gvisor v0.0.0-20250428193742-2d800c3129d5
h12.io/socks v1.0.3 h12.io/socks v1.0.3
lukechampine.com/blake3 v1.4.1 lukechampine.com/blake3 v1.4.1
+2 -2
View File
@@ -145,8 +145,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 h1:
google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
google.golang.org/grpc v1.75.0 h1:+TW+dqTd2Biwe6KKfhE5JpiYIBWq865PhKGSXiivqt4= google.golang.org/grpc v1.75.0 h1:+TW+dqTd2Biwe6KKfhE5JpiYIBWq865PhKGSXiivqt4=
google.golang.org/grpc v1.75.0/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= google.golang.org/grpc v1.75.0/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
google.golang.org/protobuf v1.36.7 h1:IgrO7UwFQGJdRNXH/sQux4R1Dj1WAKcLElzeeRaXV2A= google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
google.golang.org/protobuf v1.36.7/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+28 -16
View File
@@ -5,6 +5,7 @@ import (
go_errors "errors" go_errors "errors"
"io" "io"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/xtls/xray-core/common" "github.com/xtls/xray-core/common"
@@ -369,6 +370,10 @@ type outboundConn struct {
access sync.Mutex access sync.Mutex
dialer func() (stat.Connection, error) dialer func() (stat.Connection, error)
closeOnce sync.Once
dialOnce sync.Once
closed atomic.Bool
conn net.Conn conn net.Conn
connReady chan struct{} connReady chan struct{}
} }
@@ -378,24 +383,28 @@ func (c *outboundConn) dial() error {
if err != nil { if err != nil {
return err return err
} }
if c.closed.Load() {
return errors.New("connection closed during dial")
}
c.conn = conn c.conn = conn
c.connReady <- struct{}{} c.connReady <- struct{}{}
return nil return nil
} }
func (c *outboundConn) Write(b []byte) (int, error) { func (c *outboundConn) Write(b []byte) (int, error) {
c.dialOnce.Do(func() {
c.dial()
})
c.access.Lock() c.access.Lock()
conn := c.conn
if c.conn == nil {
if err := c.dial(); err != nil {
c.access.Unlock()
errors.LogWarningInner(context.Background(), err, "failed to dial outbound connection")
return len(b), nil
}
}
c.access.Unlock() c.access.Unlock()
if conn == nil {
_, open := <-c.connReady
if !open {
return 0, io.EOF
}
conn = c.conn
}
return c.conn.Write(b) return c.conn.Write(b)
} }
@@ -417,11 +426,14 @@ func (c *outboundConn) Read(b []byte) (int, error) {
} }
func (c *outboundConn) Close() error { func (c *outboundConn) Close() error {
c.access.Lock() c.closeOnce.Do(func() {
close(c.connReady) c.access.Lock()
if c.conn != nil { c.closed.Store(true)
c.conn.Close() close(c.connReady)
} if c.conn != nil {
c.access.Unlock() c.conn.Close()
}
c.access.Unlock()
})
return nil return nil
} }
+58 -16
View File
@@ -18,6 +18,7 @@ import (
"github.com/xtls/xray-core/common/session" "github.com/xtls/xray-core/common/session"
"github.com/xtls/xray-core/common/signal" "github.com/xtls/xray-core/common/signal"
"github.com/xtls/xray-core/common/task" "github.com/xtls/xray-core/common/task"
"github.com/xtls/xray-core/common/utils"
"github.com/xtls/xray-core/core" "github.com/xtls/xray-core/core"
"github.com/xtls/xray-core/features/policy" "github.com/xtls/xray-core/features/policy"
"github.com/xtls/xray-core/features/stats" "github.com/xtls/xray-core/features/stats"
@@ -72,7 +73,7 @@ func isValidAddress(addr *net.IPOrDomain) bool {
} }
a := addr.AsAddress() a := addr.AsAddress()
return a != net.AnyIP && a != net.AnyIPv6 return a != net.AnyIP
} }
// Process implements proxy.Outbound. // Process implements proxy.Outbound.
@@ -189,7 +190,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
writer = buf.NewWriter(conn) writer = buf.NewWriter(conn)
} }
} else { } else {
writer = NewPacketWriter(conn, UDPOverride, destination) writer = NewPacketWriter(conn, h, UDPOverride, destination)
if h.config.Noises != nil { if h.config.Noises != nil {
errors.LogDebug(ctx, "NOISE", h.config.Noises) errors.LogDebug(ctx, "NOISE", h.config.Noises)
writer = &NoisePacketWriter{ writer = &NoisePacketWriter{
@@ -261,7 +262,7 @@ func isTLSConn(conn stat.Connection) bool {
return false return false
} }
func NewPacketReader(conn net.Conn, UDPOverride net.Destination, destination net.Destination) buf.Reader { func NewPacketReader(conn net.Conn, UDPOverride net.Destination, DialDest net.Destination) buf.Reader {
iConn := conn iConn := conn
statConn, ok := iConn.(*stat.CounterConnection) statConn, ok := iConn.(*stat.CounterConnection)
if ok { if ok {
@@ -281,7 +282,7 @@ func NewPacketReader(conn net.Conn, UDPOverride net.Destination, destination net
PacketConnWrapper: c, PacketConnWrapper: c,
Counter: counter, Counter: counter,
IsOverridden: isOverridden, IsOverridden: isOverridden,
InitUnchangedAddr: destination.Address, InitUnchangedAddr: DialDest.Address,
InitChangedAddr: net.DestinationFromAddr(conn.RemoteAddr()).Address, InitChangedAddr: net.DestinationFromAddr(conn.RemoteAddr()).Address,
} }
} }
@@ -325,7 +326,7 @@ func (r *PacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
} }
// DialDest means the dial target used in the dialer when creating conn // DialDest means the dial target used in the dialer when creating conn
func NewPacketWriter(conn net.Conn, UDPOverride net.Destination, destination net.Destination) buf.Writer { func NewPacketWriter(conn net.Conn, h *Handler, UDPOverride net.Destination, DialDest net.Destination) buf.Writer {
iConn := conn iConn := conn
statConn, ok := iConn.(*stat.CounterConnection) statConn, ok := iConn.(*stat.CounterConnection)
if ok { if ok {
@@ -336,12 +337,19 @@ func NewPacketWriter(conn net.Conn, UDPOverride net.Destination, destination net
counter = statConn.WriteCounter counter = statConn.WriteCounter
} }
if c, ok := iConn.(*internet.PacketConnWrapper); ok { if c, ok := iConn.(*internet.PacketConnWrapper); ok {
// If DialDest is a domain, it will be resolved in dialer
// check this behavior and add it to map
resolvedUDPAddr := utils.NewTypedSyncMap[string, net.Address]()
if DialDest.Address.Family().IsDomain() {
resolvedUDPAddr.Store(DialDest.Address.Domain(), net.DestinationFromAddr(conn.RemoteAddr()).Address)
}
return &PacketWriter{ return &PacketWriter{
PacketConnWrapper: c, PacketConnWrapper: c,
Counter: counter, Counter: counter,
Handler: h,
UDPOverride: UDPOverride, UDPOverride: UDPOverride,
InitUnchangedAddr: destination.Address, ResolvedUDPAddr: resolvedUDPAddr,
InitChangedAddr: net.DestinationFromAddr(conn.RemoteAddr()).Address, LocalAddr: net.DestinationFromAddr(conn.LocalAddr()).Address,
} }
} }
@@ -351,10 +359,15 @@ func NewPacketWriter(conn net.Conn, UDPOverride net.Destination, destination net
type PacketWriter struct { type PacketWriter struct {
*internet.PacketConnWrapper *internet.PacketConnWrapper
stats.Counter stats.Counter
*Handler
UDPOverride net.Destination UDPOverride net.Destination
InitUnchangedAddr net.Address // Dest of udp packets might be a domain, we will resolve them to IP
InitChangedAddr net.Address // But resolver will return a random one if the domain has many IPs
// Resulting in these packets being sent to many different IPs randomly
// So, cache and keep the resolve result
ResolvedUDPAddr *utils.TypedSyncMap[string, net.Address]
LocalAddr net.Address
} }
func (w *PacketWriter) WriteMultiBuffer(mb buf.MultiBuffer) error { func (w *PacketWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
@@ -373,15 +386,44 @@ func (w *PacketWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
if w.UDPOverride.Port != 0 { if w.UDPOverride.Port != 0 {
b.UDP.Port = w.UDPOverride.Port b.UDP.Port = w.UDPOverride.Port
} }
if b.UDP.Address == w.InitUnchangedAddr {
b.UDP.Address = w.InitChangedAddr
}
if b.UDP.Address.Family().IsDomain() { if b.UDP.Address.Family().IsDomain() {
b.Release() if ip, ok := w.ResolvedUDPAddr.Load(b.UDP.Address.Domain()); ok {
buf.ReleaseMulti(mb) b.UDP.Address = ip
return errors.New("multiple domains cone is not supported") } else {
ShouldUseSystemResolver := true
if w.Handler.config.DomainStrategy.HasStrategy() {
ips, err := internet.LookupForIP(b.UDP.Address.Domain(), w.Handler.config.DomainStrategy, w.LocalAddr)
if err != nil {
// drop packet if resolve failed when forceIP
if w.Handler.config.DomainStrategy.ForceIP() {
b.Release()
continue
}
} else {
ip = net.IPAddress(ips[dice.Roll(len(ips))])
ShouldUseSystemResolver = false
}
}
if ShouldUseSystemResolver {
udpAddr, err := net.ResolveUDPAddr("udp", b.UDP.NetAddr())
if err != nil {
b.Release()
continue
} else {
ip = net.IPAddress(udpAddr.IP)
}
}
if ip != nil {
b.UDP.Address, _ = w.ResolvedUDPAddr.LoadOrStore(b.UDP.Address.Domain(), ip)
}
}
} }
n, err = w.PacketConnWrapper.WriteTo(b.Bytes(), b.UDP.RawNetAddr()) destAddr, _ := net.ResolveUDPAddr("udp", b.UDP.NetAddr())
if destAddr == nil {
b.Release()
continue
}
n, err = w.PacketConnWrapper.WriteTo(b.Bytes(), destAddr)
} else { } else {
n, err = w.PacketConnWrapper.Write(b.Bytes()) n, err = w.PacketConnWrapper.Write(b.Bytes())
} }