detail

proxy/shadowsocks_2022/inbound.go

@@ -169,17 +169,11 @@ func (i *Inbound) processUDP(ctx context.Context, conn stat.Connection, dispatch
 				}
 				sessionItem.Unlock()
 			}
-			link := sessionItem.link.Load()
+			link, err := sessionItem.EnsureLink(ctx, conn, decoded.Destination, dispatcher, i.policyManager, func(dest net.Destination, payload []byte) ([]byte, error) {
-			if link != nil {
+				return i.udpCodec.EncodeServerPacket(decoded.SessionID, dest, payload)
-				sessionItem.UpdateConn(conn)
+			})
-			} else {
+			if err != nil {
-				var err error
+				continue
-				link, err = sessionItem.EnsureLink(ctx, conn, decoded.Destination, dispatcher, i.policyManager, func(dest net.Destination, payload []byte) ([]byte, error) {
-					return i.udpCodec.EncodeServerPacket(decoded.SessionID, dest, payload)
-				})
-				if err != nil {
-					continue
-				}
 			}
 
 			payloadBuf := buf.New()

proxy/shadowsocks_2022/inbound_multi.go

@@ -336,17 +336,11 @@ func (i *MultiUserInbound) processUDP(ctx context.Context, conn stat.Connection,
 			}
 			sessionItem.Unlock()
 
-			link := sessionItem.link.Load()
+			link, err := sessionItem.EnsureLink(ctx, conn, decoded.Destination, dispatcher, i.policyManager, func(replyDest net.Destination, payload []byte) ([]byte, error) {
-			if link != nil {
+				return i.encodeServerUDPPacket(sessionID, userPSK, replyDest, payload)
-				sessionItem.UpdateConn(conn)
+			})
-			} else {
+			if err != nil {
-				var err error
+				continue
-				link, err = sessionItem.EnsureLink(ctx, conn, decoded.Destination, dispatcher, i.policyManager, func(replyDest net.Destination, payload []byte) ([]byte, error) {
-					return i.encodeServerUDPPacket(sessionID, userPSK, replyDest, payload)
-				})
-				if err != nil {
-					continue
-				}
 			}
 
 			pBuf := buf.New()

proxy/shadowsocks_2022/inbound_relay.go

@@ -253,16 +253,10 @@ func (i *RelayInbound) processUDP(ctx context.Context, conn stat.Connection, dis
 				}
 				sessionItem.Unlock()
 			}
-			link := sessionItem.link.Load()
+			link, err := sessionItem.EnsureLink(ctx, conn, dest, dispatcher, i.policyManager, nil)
-			if link != nil {
+			if err != nil {
-				sessionItem.UpdateConn(conn)
+				b.Release()
-			} else {
+				continue
-				var err error
-				link, err = sessionItem.EnsureLink(ctx, conn, dest, dispatcher, i.policyManager, nil)
-				if err != nil {
-					b.Release()
-					continue
-				}
 			}
 
 			_ = link.Writer.WriteMultiBuffer(buf.MultiBuffer{b})

proxy/shadowsocks_2022/outbound.go

@@ -169,8 +169,8 @@ func (o *Outbound) Process(ctx context.Context, link *transport.Link, dialer int
 			return buf.Copy(responseReader, link.Writer, buf.UpdateActivity(timer))
 		}
 
-		responseDonePost := task.OnSuccess(responseDone, task.Close(link.Writer))
+		responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer))
-		if err := task.Run(ctx, requestDone, responseDonePost); err != nil {
+		if err := task.Run(ctx, requestDone, responseDoneAndCloseWriter); err != nil {
 			return errors.New("connection ends").Base(err)
 		}
 
@@ -212,8 +212,8 @@ func (o *Outbound) Process(ctx context.Context, link *transport.Link, dialer int
 			return nil
 		}
 
-		responseDonePost := task.OnSuccess(responseDone, task.Close(link.Writer))
+		responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer))
-		if err := task.Run(ctx, requestDone, responseDonePost); err != nil {
+		if err := task.Run(ctx, requestDone, responseDoneAndCloseWriter); err != nil {
 			return errors.New("connection ends").Base(err)
 		}
 

proxy/shadowsocks_2022/shadowsocks_2022.go

@@ -18,8 +18,12 @@ import (
 )
 
 func (s *ServerUDPSession) UpdateConn(conn stat.Connection) {
-	s.currentConn.Store(conn)
+	if s.currentConn.Load() == nil {
-	s.timer.Update()
+		s.currentConn.Store(conn)
+	}
+	if s.timer != nil {
+		s.timer.Update()
+	}
 }
 
 func (s *ServerUDPSession) WriteToClient(b []byte) error {
@@ -53,16 +57,19 @@ func (s *ServerUDPSession) EnsureLink(
 	policyManager policy.Manager,
 	responseEncoder func(dest net.Destination, payload []byte) ([]byte, error),
 ) (*transport.Link, error) {
+	s.UpdateConn(conn)
+
+	if link := s.link.Load(); link != nil {
+		return link, nil
+	}
+
 	s.Lock()
 	defer s.Unlock()
 
 	if link := s.link.Load(); link != nil {
-		s.UpdateConn(conn)
 		return link, nil
 	}
 
-	s.currentConn.Store(conn)
-
 	sessCtx, cancel := context.WithCancel(ctx)
 	inbound := session.InboundFromContext(sessCtx)
 	if inbound != nil && s.User != nil {
@@ -87,6 +94,7 @@ func (s *ServerUDPSession) EnsureLink(
 		return nil, err
 	}
 
+	s.link.Store(link)
 	sessionPolicy := policyManager.ForLevel(level)
 	s.timer = signal.CancelAfterInactivity(sessCtx, func() {
 		if s.manager != nil {
@@ -95,7 +103,6 @@ func (s *ServerUDPSession) EnsureLink(
 		s.Close()
 		cancel()
 	}, sessionPolicy.Timeouts.ConnectionIdle)
-	s.link.Store(link)
 
 	go handleUDPResponse(s, link, dest, responseEncoder)
 	return link, nil
@@ -111,13 +118,19 @@ func ResetTCPConn(conn net.Conn) {
 }
 
 func handleUDPResponse(s *ServerUDPSession, link *transport.Link, fallbackDest net.Destination, encode func(dest net.Destination, payload []byte) ([]byte, error)) {
-	defer s.timer.SetTimeout(0)
+	defer func() {
+		if s.timer != nil {
+			s.timer.SetTimeout(0)
+		}
+	}()
 	for {
 		resMb, err := link.Reader.ReadMultiBuffer()
 		if err != nil {
 			return
 		}
-		s.timer.Update()
+		if s.timer != nil {
+			s.timer.Update()
+		}
 		for i, rb := range resMb {
 			b := rb.Bytes()
 			if encode != nil {

proxy/shadowsocks_2022/stream.go

@@ -11,7 +11,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/antireplay"
 	"github.com/xtls/xray-core/common/buf"
 	"github.com/xtls/xray-core/common/errors"
@@ -630,7 +629,6 @@ func InitServerStream(conn net.Conn, method *CipherMethod, psk, saltSlice []byte
 func TransportTCP(ctx context.Context, sessionPolicy policy.Session, reader buf.Reader, writer buf.Writer, link *transport.Link) error {
 	ctx, cancel := context.WithCancel(ctx)
 	timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
-	defer timer.SetTimeout(0)
 	ctx = policy.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
 
 	requestDone := func() error {
@@ -646,12 +644,6 @@ func TransportTCP(ctx context.Context, sessionPolicy policy.Session, reader buf.
 		return buf.Copy(link.Reader, writer, buf.UpdateActivity(timer))
 	}
 
-	requestDoneAndCloseWriter := task.OnSuccess(requestDone, task.Close(link.Writer))
+	responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer))
-	if err := task.Run(ctx, requestDoneAndCloseWriter, responseDone); err != nil {
+	return task.Run(ctx, requestDone, responseDoneAndCloseWriter)
-		common.Interrupt(link.Reader)
-		common.Interrupt(link.Writer)
-		return errors.New("connection ends").Base(err)
-	}
-
-	return nil
 }

proxy/tun/handler.go

@@ -125,7 +125,8 @@ func (t *Handler) Start() error {
 				errors.LogInfo(context.Background(), "[tun] falied to set interface > iface == nil")
 				return errors.New("iface not found")
 			}
-			return c.Control(func(fd uintptr) {
+			var controlErr, sockErr error
+			controlErr = c.Control(func(fd uintptr) {
 				addrPort, _ := netip.ParseAddrPort(address)
 				// skip loopback
 				if addrPort.Addr().IsLoopback() || strings.HasPrefix(strings.ToLower(address), "localhost:") {
@@ -133,9 +134,13 @@ func (t *Handler) Start() error {
 				}
 				err := setinterface(network, address, fd, iface)
 				if err != nil {
-					errors.LogInfoInner(context.Background(), err, "[tun] falied to set interface")
+					sockErr = errors.New("[tun] falied to set interface").Base(err)
 				}
 			})
+			if sockErr != nil {
+				return sockErr
+			}
+			return controlErr
 		})
 	}
 

transport/internet/sockopt.go

@@ -28,3 +28,22 @@ func (v *SocketConfig) ParseTFOValue() int {
 	}
 	return tfo
 }
+
+// Dial should fail on bind errors, return this to indicate such errors.
+type BindError struct {
+	err error
+}
+
+func (e *BindError) Unwrap() error {
+	return e.err
+}
+
+func (e *BindError) Error() string {
+	return "failed to bind to interface: " + e.err.Error()
+}
+
+func NewBindError(err error) *BindError {
+	return &BindError{
+		err: err,
+	}
+}

transport/internet/sockopt_darwin.go

@@ -136,15 +136,15 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 	if config.Interface != "" {
 		iface, err := net.InterfaceByName(config.Interface)
 		if err != nil {
-			return errors.New("failed to get interface ", config.Interface).Base(err)
+			return NewBindError(errors.New("failed to get interface ", config.Interface).Base(err))
 		}
 		if network == "tcp6" || network == "udp6" {
 			if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF, iface.Index); err != nil {
-				return errors.New("failed to set IPV6_BOUND_IF").Base(err)
+				return NewBindError(errors.New("failed to set IPV6_BOUND_IF").Base(err))
 			}
 		} else {
 			if err := unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_BOUND_IF, iface.Index); err != nil {
-				return errors.New("failed to set IP_BOUND_IF").Base(err)
+				return NewBindError(errors.New("failed to set IP_BOUND_IF").Base(err))
 			}
 		}
 	}

transport/internet/sockopt_linux.go

@@ -22,7 +22,7 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 
 	if config.Interface != "" {
 		if err := syscall.BindToDevice(int(fd), config.Interface); err != nil {
-			return errors.New("failed to set Interface").Base(err)
+			return NewBindError(errors.New("failed to set Interface").Base(err))
 		}
 	}
 

transport/internet/sockopt_windows.go

@@ -35,7 +35,7 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 	if config.Interface != "" {
 		inf, err := net.InterfaceByName(config.Interface)
 		if err != nil {
-			return errors.New("failed to find the interface").Base(err)
+			return NewBindError(errors.New("failed to find the interface").Base(err))
 		}
 		// easy way to check if the address is ipv4
 		isV4 := strings.Contains(address, ".")
@@ -47,20 +47,20 @@ func applyOutboundSocketOptions(network string, address string, fd uintptr, conf
 			binary.BigEndian.PutUint32(bytes[:], uint32(inf.Index))
 			idx := *(*uint32)(unsafe.Pointer(&bytes[0]))
 			if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IP, IP_UNICAST_IF, int(idx)); err != nil {
-				return errors.New("failed to set IP_UNICAST_IF").Base(err)
+				return NewBindError(errors.New("failed to set IP_UNICAST_IF").Base(err))
 			}
 			if ip := net.ParseIP(host); ip != nil && ip.IsMulticast() && isUDPSocket(network) {
 				if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IP, syscall.IP_MULTICAST_IF, int(idx)); err != nil {
-					return errors.New("failed to set IP_MULTICAST_IF").Base(err)
+					return NewBindError(errors.New("failed to set IP_MULTICAST_IF").Base(err))
 				}
 			}
 		} else {
 			if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IPV6, IPV6_UNICAST_IF, inf.Index); err != nil {
-				return errors.New("failed to set IPV6_UNICAST_IF").Base(err)
+				return NewBindError(errors.New("failed to set IPV6_UNICAST_IF").Base(err))
 			}
 			if ip := net.ParseIP(host); ip != nil && ip.IsMulticast() && isUDPSocket(network) {
 				if err := syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_IPV6, syscall.IPV6_MULTICAST_IF, inf.Index); err != nil {
-					return errors.New("failed to set IPV6_MULTICAST_IF").Base(err)
+					return NewBindError(errors.New("failed to set IPV6_MULTICAST_IF").Base(err))
 				}
 			}
 		}

transport/internet/system_dialer.go

@@ -68,19 +68,27 @@ func (d *DefaultSystemDialer) Dial(ctx context.Context, src net.Address, dest ne
 			}
 		}
 		var lc net.ListenConfig
-		lc.Control = func(network, address string, c syscall.RawConn) error {
+		if sockopt != nil || len(Controllers) > 0 {
-			for _, ctl := range Controllers {
+			lc.Control = func(network, address string, c syscall.RawConn) error {
-				if err := ctl(network, address, c); err != nil {
+				for _, ctl := range Controllers {
-					errors.LogInfoInner(ctx, err, "failed to apply external controller")
+					if err := ctl(network, address, c); err != nil {
-				}
+						return errors.New("failed to apply external controller").Base(err)
-			}
-			return c.Control(func(fd uintptr) {
-				if sockopt != nil {
-					if err := applyOutboundSocketOptions(network, destAddr.String(), fd, sockopt); err != nil {
-						errors.LogInfo(ctx, err, "failed to apply socket options")
 					}
 				}
-			})
+				if sockopt != nil {
+					var sockErr, controlErr error
+					controlErr = c.Control(func(fd uintptr) {
+						if sockErr = applyOutboundSocketOptions(network, destAddr.String(), fd, sockopt); sockErr != nil {
+							errors.LogInfoInner(ctx, sockErr, "failed to apply socket options")
+						}
+					})
+					if _, ok := sockErr.(*BindError); ok {
+						return sockErr
+					}
+					return controlErr
+				}
+				return nil
+			}
 		}
 		packetConn, err := lc.ListenPacket(ctx, srcAddr.Network(), srcAddr.String())
 		if err != nil {
@@ -128,16 +136,22 @@ func (d *DefaultSystemDialer) Dial(ctx context.Context, src net.Address, dest ne
 		dialer.Control = func(network, address string, c syscall.RawConn) error {
 			for _, ctl := range Controllers {
 				if err := ctl(network, address, c); err != nil {
-					errors.LogInfoInner(ctx, err, "failed to apply external controller")
+					return errors.New("failed to apply external controller").Base(err)
 				}
 			}
-			return c.Control(func(fd uintptr) {
+			if sockopt != nil {
-				if sockopt != nil {
+				var sockErr, controlErr error
-					if err := applyOutboundSocketOptions(network, address, fd, sockopt); err != nil {
+				controlErr = c.Control(func(fd uintptr) {
-						errors.LogInfoInner(ctx, err, "failed to apply socket options")
+					if sockErr = applyOutboundSocketOptions(network, address, fd, sockopt); sockErr != nil {
+						errors.LogInfoInner(ctx, sockErr, "failed to apply socket options")
 					}
+				})
+				if _, ok := sockErr.(*BindError); ok {
+					return sockErr
 				}
-			})
+				return controlErr
+			}
+			return nil
 		}
 	}