Crazy sekai overturns the small pond

inbound/default.go

@@ -22,13 +22,13 @@ type myInboundAdapter struct {
 	protocol         string
 	network          []string
 	ctx              context.Context
-	router           adapter.ConnectionRouter
+	router           adapter.ConnectionRouterEx
 	logger           log.ContextLogger
 	tag              string
 	listenOptions    option.ListenOptions
-	connHandler      adapter.ConnectionHandler
-	packetHandler    adapter.PacketHandler
-	oobPacketHandler adapter.OOBPacketHandler
+	connHandler      adapter.ConnectionHandlerEx
+	packetHandler    adapter.PacketHandlerEx
+	oobPacketHandler adapter.OOBPacketHandlerEx
 	packetUpstream   any
 
 	// http mixed
@@ -55,10 +55,6 @@ func (a *myInboundAdapter) Tag() string {
 	return a.tag
 }
 
-func (a *myInboundAdapter) Network() []string {
-	return a.network
-}
-
 func (a *myInboundAdapter) Start() error {
 	var err error
 	if common.Contains(a.network, N.NetworkTCP) {
@@ -150,6 +146,31 @@ func (a *myInboundAdapter) newPacketConnection(ctx context.Context, conn N.Packe
 	return a.router.RoutePacketConnection(ctx, conn, metadata)
 }
 
+func (a *myInboundAdapter) upstreamHandlerEx(metadata adapter.InboundContext) adapter.UpstreamHandlerAdapterEx {
+	return adapter.NewUpstreamHandlerEx(metadata, a.newConnectionEx, a.streamPacketConnectionEx)
+}
+
+func (a *myInboundAdapter) upstreamContextHandlerEx() adapter.UpstreamHandlerAdapterEx {
+	return adapter.NewUpstreamContextHandlerEx(a.newConnectionEx, a.newPacketConnectionEx)
+}
+
+func (a *myInboundAdapter) newConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	a.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
+	a.router.RouteConnectionEx(ctx, conn, metadata, onClose)
+}
+
+func (a *myInboundAdapter) newPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	ctx = log.ContextWithNewID(ctx)
+	a.logger.InfoContext(ctx, "inbound packet connection from ", metadata.Source)
+	a.logger.InfoContext(ctx, "inbound packet connection to ", metadata.Destination)
+	a.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
+}
+
+func (a *myInboundAdapter) streamPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	a.logger.InfoContext(ctx, "inbound packet connection to ", metadata.Destination)
+	a.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
+}
+
 func (a *myInboundAdapter) createMetadata(conn net.Conn, metadata adapter.InboundContext) adapter.InboundContext {
 	metadata.Inbound = a.tag
 	metadata.InboundType = a.protocol
@@ -167,25 +188,17 @@ func (a *myInboundAdapter) createMetadata(conn net.Conn, metadata adapter.Inboun
 	return metadata
 }
 
-func (a *myInboundAdapter) createPacketMetadata(conn N.PacketConn, metadata adapter.InboundContext) adapter.InboundContext {
-	metadata.Inbound = a.tag
-	metadata.InboundType = a.protocol
-	metadata.InboundDetour = a.listenOptions.Detour
-	metadata.InboundOptions = a.listenOptions.InboundOptions
-	if !metadata.Destination.IsValid() {
-		metadata.Destination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap()
-	}
-	return metadata
-}
-
+// Deprecated: don't use
 func (a *myInboundAdapter) newError(err error) {
 	a.logger.Error(err)
 }
 
+// Deprecated: don't use
 func (a *myInboundAdapter) NewError(ctx context.Context, err error) {
 	NewError(a.logger, ctx, err)
 }
 
+// Deprecated: don't use
 func NewError(logger log.ContextLogger, ctx context.Context, err error) {
 	common.Close(err)
 	if E.IsClosedOrCanceled(err) {

inbound/default_tcp.go

@@ -71,18 +71,14 @@ func (a *myInboundAdapter) injectTCP(conn net.Conn, metadata adapter.InboundCont
 	ctx := log.ContextWithNewID(a.ctx)
 	metadata = a.createMetadata(conn, metadata)
 	a.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
-	hErr := a.connHandler.NewConnection(ctx, conn, metadata)
-	if hErr != nil {
-		conn.Close()
-		a.NewError(ctx, E.Cause(hErr, "process connection from ", metadata.Source))
-	}
+	a.connHandler.NewConnectionEx(ctx, conn, metadata, nil)
 }
 
-func (a *myInboundAdapter) routeTCP(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) {
+func (a *myInboundAdapter) routeTCP(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	metadata := a.createMetadata(conn, adapter.InboundContext{
+		Source:      source,
+		Destination: destination,
+	})
 	a.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
-	hErr := a.newConnection(ctx, conn, metadata)
-	if hErr != nil {
-		conn.Close()
-		a.NewError(ctx, E.Cause(hErr, "process connection from ", metadata.Source))
-	}
+	a.connHandler.NewConnectionEx(ctx, conn, metadata, onClose)
 }

inbound/default_udp.go

@@ -42,7 +42,6 @@ func (a *myInboundAdapter) loopUDPIn() {
 	defer buffer.Release()
 	buffer.IncRef()
 	defer buffer.DecRef()
-	packetService := (*myInboundPacketAdapter)(a)
 	for {
 		buffer.Reset()
 		n, addr, err := a.udpConn.ReadFromUDPAddrPort(buffer.FreeBytes())
@@ -50,16 +49,7 @@ func (a *myInboundAdapter) loopUDPIn() {
 			return
 		}
 		buffer.Truncate(n)
-		var metadata adapter.InboundContext
-		metadata.Inbound = a.tag
-		metadata.InboundType = a.protocol
-		metadata.InboundOptions = a.listenOptions.InboundOptions
-		metadata.Source = M.SocksaddrFromNetIP(addr).Unwrap()
-		metadata.OriginDestination = a.udpAddr
-		err = a.packetHandler.NewPacket(a.ctx, packetService, buffer, metadata)
-		if err != nil {
-			a.newError(E.Cause(err, "process packet from ", metadata.Source))
-		}
+		a.packetHandler.NewPacketEx(buffer, M.SocksaddrFromNetIP(addr).Unwrap())
 	}
 }
 
@@ -69,7 +59,6 @@ func (a *myInboundAdapter) loopUDPOOBIn() {
 	defer buffer.Release()
 	buffer.IncRef()
 	defer buffer.DecRef()
-	packetService := (*myInboundPacketAdapter)(a)
 	oob := make([]byte, 1024)
 	for {
 		buffer.Reset()
@@ -78,22 +67,12 @@ func (a *myInboundAdapter) loopUDPOOBIn() {
 			return
 		}
 		buffer.Truncate(n)
-		var metadata adapter.InboundContext
-		metadata.Inbound = a.tag
-		metadata.InboundType = a.protocol
-		metadata.InboundOptions = a.listenOptions.InboundOptions
-		metadata.Source = M.SocksaddrFromNetIP(addr).Unwrap()
-		metadata.OriginDestination = a.udpAddr
-		err = a.oobPacketHandler.NewPacket(a.ctx, packetService, buffer, oob[:oobN], metadata)
-		if err != nil {
-			a.newError(E.Cause(err, "process packet from ", metadata.Source))
-		}
+		a.oobPacketHandler.NewPacketEx(buffer, oob[:oobN], M.SocksaddrFromNetIP(addr).Unwrap())
 	}
 }
 
 func (a *myInboundAdapter) loopUDPInThreadSafe() {
 	defer close(a.packetOutboundClosed)
-	packetService := (*myInboundPacketAdapter)(a)
 	for {
 		buffer := buf.NewPacket()
 		n, addr, err := a.udpConn.ReadFromUDPAddrPort(buffer.FreeBytes())
@@ -102,23 +81,12 @@ func (a *myInboundAdapter) loopUDPInThreadSafe() {
 			return
 		}
 		buffer.Truncate(n)
-		var metadata adapter.InboundContext
-		metadata.Inbound = a.tag
-		metadata.InboundType = a.protocol
-		metadata.InboundOptions = a.listenOptions.InboundOptions
-		metadata.Source = M.SocksaddrFromNetIP(addr).Unwrap()
-		metadata.OriginDestination = a.udpAddr
-		err = a.packetHandler.NewPacket(a.ctx, packetService, buffer, metadata)
-		if err != nil {
-			buffer.Release()
-			a.newError(E.Cause(err, "process packet from ", metadata.Source))
-		}
+		a.packetHandler.NewPacketEx(buffer, M.SocksaddrFromNetIP(addr).Unwrap())
 	}
 }
 
 func (a *myInboundAdapter) loopUDPOOBInThreadSafe() {
 	defer close(a.packetOutboundClosed)
-	packetService := (*myInboundPacketAdapter)(a)
 	oob := make([]byte, 1024)
 	for {
 		buffer := buf.NewPacket()
@@ -128,17 +96,7 @@ func (a *myInboundAdapter) loopUDPOOBInThreadSafe() {
 			return
 		}
 		buffer.Truncate(n)
-		var metadata adapter.InboundContext
-		metadata.Inbound = a.tag
-		metadata.InboundType = a.protocol
-		metadata.InboundOptions = a.listenOptions.InboundOptions
-		metadata.Source = M.SocksaddrFromNetIP(addr).Unwrap()
-		metadata.OriginDestination = a.udpAddr
-		err = a.oobPacketHandler.NewPacket(a.ctx, packetService, buffer, oob[:oobN], metadata)
-		if err != nil {
-			buffer.Release()
-			a.newError(E.Cause(err, "process packet from ", metadata.Source))
-		}
+		a.oobPacketHandler.NewPacketEx(buffer, oob[:oobN], M.SocksaddrFromNetIP(addr).Unwrap())
 	}
 }
 
@@ -148,7 +106,7 @@ func (a *myInboundAdapter) loopUDPOut() {
 		case packet := <-a.packetOutbound:
 			err := a.writePacket(packet.buffer, packet.destination)
 			if err != nil && !E.IsClosed(err) {
-				a.newError(E.New("write back udp: ", err))
+				a.logger.Error(E.New("write back udp: ", err))
 			}
 			continue
 		case <-a.packetOutboundClosed:
@@ -164,15 +122,36 @@ func (a *myInboundAdapter) loopUDPOut() {
 	}
 }
 
+func (a *myInboundAdapter) packetConn() N.PacketConn {
+	return (*myInboundPacketAdapter)(a)
+}
+
+func (a *myInboundAdapter) createPacketMetadata(conn N.PacketConn, metadata adapter.InboundContext) adapter.InboundContext {
+	metadata.Inbound = a.tag
+	metadata.InboundType = a.protocol
+	metadata.InboundDetour = a.listenOptions.Detour
+	metadata.InboundOptions = a.listenOptions.InboundOptions
+	if !metadata.Destination.IsValid() {
+		metadata.Destination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap()
+	}
+	metadata.OriginDestination = a.udpAddr
+	return metadata
+}
+
+func (a *myInboundAdapter) createPacketMetadataEx(source M.Socksaddr, destination M.Socksaddr) adapter.InboundContext {
+	var metadata adapter.InboundContext
+	metadata.Inbound = a.tag
+	metadata.InboundType = a.protocol
+	metadata.InboundDetour = a.listenOptions.Detour
+	metadata.InboundOptions = a.listenOptions.InboundOptions
+	metadata.Source = source
+	metadata.Destination = destination
+	metadata.OriginDestination = a.udpAddr
+	return metadata
+}
+
 func (a *myInboundAdapter) writePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
 	defer buffer.Release()
-	if destination.IsFqdn() {
-		udpAddr, err := net.ResolveUDPAddr(N.NetworkUDP, destination.String())
-		if err != nil {
-			return err
-		}
-		return common.Error(a.udpConn.WriteTo(buffer.Bytes(), udpAddr))
-	}
 	return common.Error(a.udpConn.WriteToUDPAddrPort(buffer.Bytes(), destination.AddrPort()))
 }
 

inbound/direct.go

@@ -3,7 +3,6 @@ package inbound
 import (
 	"context"
 	"net"
-	"net/netip"
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
@@ -13,14 +12,14 @@ import (
 	"github.com/sagernet/sing/common/buf"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
-	"github.com/sagernet/sing/common/udpnat"
+	"github.com/sagernet/sing/common/udpnat2"
 )
 
 var _ adapter.Inbound = (*Direct)(nil)
 
 type Direct struct {
 	myInboundAdapter
-	udpNat              *udpnat.Service[netip.AddrPort]
+	udpNat              *udpnat.Service
 	overrideOption      int
 	overrideDestination M.Socksaddr
 }
@@ -54,10 +53,9 @@ func NewDirect(ctx context.Context, router adapter.Router, logger log.ContextLog
 	} else {
 		udpTimeout = C.UDPTimeout
 	}
-	inbound.udpNat = udpnat.New[netip.AddrPort](int64(udpTimeout.Seconds()), adapter.NewUpstreamContextHandler(inbound.newConnection, inbound.newPacketConnection, inbound))
+	inbound.udpNat = udpnat.New(inbound, inbound.preparePacketConnection, udpTimeout, false)
 	inbound.connHandler = inbound
 	inbound.packetHandler = inbound
-	inbound.packetUpstream = inbound.udpNat
 	return inbound
 }
 
@@ -76,29 +74,38 @@ func (d *Direct) NewConnection(ctx context.Context, conn net.Conn, metadata adap
 	return d.router.RouteConnection(ctx, conn, metadata)
 }
 
-func (d *Direct) NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, metadata adapter.InboundContext) error {
+func (d *Direct) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) {
+	var destination M.Socksaddr
 	switch d.overrideOption {
 	case 1:
-		metadata.Destination = d.overrideDestination
+		destination = d.overrideDestination
 	case 2:
-		destination := d.overrideDestination
-		destination.Port = metadata.Destination.Port
-		metadata.Destination = destination
+		destination = d.overrideDestination
+		destination.Port = source.Port
 	case 3:
-		metadata.Destination.Port = d.overrideDestination.Port
+		destination = source
+		destination.Port = d.overrideDestination.Port
 	}
-	d.udpNat.NewContextPacket(ctx, metadata.Source.AddrPort(), buffer, adapter.UpstreamMetadata(metadata), func(natConn N.PacketConn) (context.Context, N.PacketWriter) {
-		return adapter.WithContext(log.ContextWithNewID(ctx), &metadata), &udpnat.DirectBackWriter{Source: conn, Nat: natConn}
-	})
-	return nil
+	d.udpNat.NewPacket([][]byte{buffer.Bytes()}, source, destination, nil)
 }
 
-func (d *Direct) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	return d.router.RouteConnection(ctx, conn, metadata)
+func (d *Direct) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	d.newConnectionEx(ctx, conn, metadata, onClose)
 }
 
-func (d *Direct) newPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	ctx = log.ContextWithNewID(ctx)
-	d.logger.InfoContext(ctx, "inbound packet connection from ", metadata.Source)
-	return d.router.RoutePacketConnection(ctx, conn, metadata)
+func (d *Direct) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	d.newPacketConnectionEx(ctx, conn, d.createPacketMetadataEx(source, destination), onClose)
+}
+
+func (d *Direct) preparePacketConnection(source M.Socksaddr, destination M.Socksaddr, userData any) (bool, context.Context, N.PacketWriter, N.CloseHandlerFunc) {
+	return true, d.ctx, &directPacketWriter{d.packetConn(), source}, nil
+}
+
+type directPacketWriter struct {
+	writer N.PacketWriter
+	source M.Socksaddr
+}
+
+func (w *directPacketWriter) WritePacket(buffer *buf.Buffer, addr M.Socksaddr) error {
+	return w.writer.WritePacket(buffer, w.source)
 }

inbound/http.go

@@ -4,7 +4,6 @@ import (
 	std_bufio "bufio"
 	"context"
 	"net"
-	"os"
 
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/tls"
@@ -20,8 +19,8 @@ import (
 )
 
 var (
-	_ adapter.Inbound           = (*HTTP)(nil)
-	_ adapter.InjectableInbound = (*HTTP)(nil)
+	_ adapter.Inbound              = (*HTTP)(nil)
+	_ adapter.TCPInjectableInbound = (*HTTP)(nil)
 )
 
 type HTTP struct {
@@ -72,7 +71,15 @@ func (h *HTTP) Close() error {
 	)
 }
 
-func (h *HTTP) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
+func (h *HTTP) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.newConnection(ctx, conn, metadata, onClose)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
+}
+
+func (h *HTTP) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) error {
 	var err error
 	if h.tlsConfig != nil {
 		conn, err = tls.ServerHandshake(ctx, conn, h.tlsConfig)
@@ -80,35 +87,33 @@ func (h *HTTP) NewConnection(ctx context.Context, conn net.Conn, metadata adapte
 			return err
 		}
 	}
-	return http.HandleConnection(ctx, conn, std_bufio.NewReader(conn), h.authenticator, h.upstreamUserHandler(metadata), adapter.UpstreamMetadata(metadata))
+	return http.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, nil, h.upstreamUserHandlerEx(metadata), metadata.Source, onClose)
 }
 
-func (h *HTTP) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (a *myInboundAdapter) upstreamUserHandlerEx(metadata adapter.InboundContext) adapter.UpstreamHandlerAdapterEx {
+	return adapter.NewUpstreamHandlerEx(metadata, a.newUserConnection, a.streamUserPacketConnection)
 }
 
-func (a *myInboundAdapter) upstreamUserHandler(metadata adapter.InboundContext) adapter.UpstreamHandlerAdapter {
-	return adapter.NewUpstreamHandler(metadata, a.newUserConnection, a.streamUserPacketConnection, a)
-}
-
-func (a *myInboundAdapter) newUserConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
+func (a *myInboundAdapter) newUserConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
 	user, loaded := auth.UserFromContext[string](ctx)
 	if !loaded {
 		a.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
-		return a.router.RouteConnection(ctx, conn, metadata)
+		a.router.RouteConnectionEx(ctx, conn, metadata, onClose)
+		return
 	}
 	metadata.User = user
 	a.logger.InfoContext(ctx, "[", user, "] inbound connection to ", metadata.Destination)
-	return a.router.RouteConnection(ctx, conn, metadata)
+	a.router.RouteConnectionEx(ctx, conn, metadata, onClose)
 }
 
-func (a *myInboundAdapter) streamUserPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
+func (a *myInboundAdapter) streamUserPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
 	user, loaded := auth.UserFromContext[string](ctx)
 	if !loaded {
 		a.logger.InfoContext(ctx, "inbound packet connection to ", metadata.Destination)
-		return a.router.RoutePacketConnection(ctx, conn, metadata)
+		a.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
+		return
 	}
 	metadata.User = user
 	a.logger.InfoContext(ctx, "[", user, "] inbound packet connection to ", metadata.Destination)
-	return a.router.RoutePacketConnection(ctx, conn, metadata)
+	a.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
 }

inbound/mixed.go

@@ -4,7 +4,6 @@ import (
 	std_bufio "bufio"
 	"context"
 	"net"
-	"os"
 
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/uot"
@@ -12,6 +11,7 @@ import (
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
 	"github.com/sagernet/sing/common/auth"
+	E "github.com/sagernet/sing/common/exceptions"
 	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/sing/protocol/http"
 	"github.com/sagernet/sing/protocol/socks"
@@ -20,8 +20,8 @@ import (
 )
 
 var (
-	_ adapter.Inbound           = (*Mixed)(nil)
-	_ adapter.InjectableInbound = (*Mixed)(nil)
+	_ adapter.Inbound              = (*Mixed)(nil)
+	_ adapter.TCPInjectableInbound = (*Mixed)(nil)
 )
 
 type Mixed struct {
@@ -47,20 +47,24 @@ func NewMixed(ctx context.Context, router adapter.Router, logger log.ContextLogg
 	return inbound
 }
 
-func (h *Mixed) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	reader := std_bufio.NewReader(conn)
-	headerBytes, err := reader.Peek(1)
+func (h *Mixed) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.newConnection(ctx, conn, metadata, onClose)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
 	if err != nil {
-		return err
-	}
-	switch headerBytes[0] {
-	case socks4.Version, socks5.Version:
-		return socks.HandleConnection0(ctx, conn, reader, h.authenticator, h.upstreamUserHandler(metadata), adapter.UpstreamMetadata(metadata))
-	default:
-		return http.HandleConnection(ctx, conn, reader, h.authenticator, h.upstreamUserHandler(metadata), adapter.UpstreamMetadata(metadata))
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
 	}
 }
 
-func (h *Mixed) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *Mixed) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) error {
+	reader := std_bufio.NewReader(conn)
+	headerBytes, err := reader.Peek(1)
+	if err != nil {
+		return E.Cause(err, "peek first byte")
+	}
+	switch headerBytes[0] {
+	case socks4.Version, socks5.Version:
+		return socks.HandleConnectionEx(ctx, conn, reader, h.authenticator, nil, h.upstreamUserHandlerEx(metadata), metadata.Source, metadata.Destination, onClose)
+	default:
+		return http.HandleConnectionEx(ctx, conn, reader, h.authenticator, nil, h.upstreamUserHandlerEx(metadata), metadata.Source, onClose)
+	}
 }

inbound/naive.go

@@ -17,6 +17,7 @@ import (
 	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
+	"github.com/sagernet/sing-box/transport/v2rayhttp"
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/auth"
 	"github.com/sagernet/sing/common/buf"
@@ -164,13 +165,13 @@ func (n *Naive) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 			n.badRequest(ctx, request, E.New("hijack failed"))
 			return
 		}
-		n.newConnection(ctx, &naiveH1Conn{Conn: conn}, userName, source, destination)
+		n.newConnection(ctx, false, &naiveH1Conn{Conn: conn}, userName, source, destination)
 	} else {
-		n.newConnection(ctx, &naiveH2Conn{reader: request.Body, writer: writer, flusher: writer.(http.Flusher)}, userName, source, destination)
+		n.newConnection(ctx, true, &naiveH2Conn{reader: request.Body, writer: writer, flusher: writer.(http.Flusher)}, userName, source, destination)
 	}
 }
 
-func (n *Naive) newConnection(ctx context.Context, conn net.Conn, userName string, source, destination M.Socksaddr) {
+func (n *Naive) newConnection(ctx context.Context, waitForClose bool, conn net.Conn, userName string, source M.Socksaddr, destination M.Socksaddr) {
 	if userName != "" {
 		n.logger.InfoContext(ctx, "[", userName, "] inbound connection from ", source)
 		n.logger.InfoContext(ctx, "[", userName, "] inbound connection to ", destination)
@@ -178,19 +179,26 @@ func (n *Naive) newConnection(ctx context.Context, conn net.Conn, userName strin
 		n.logger.InfoContext(ctx, "inbound connection from ", source)
 		n.logger.InfoContext(ctx, "inbound connection to ", destination)
 	}
-	hErr := n.router.RouteConnection(ctx, conn, n.createMetadata(conn, adapter.InboundContext{
+	metadata := n.createMetadata(conn, adapter.InboundContext{
 		Source:      source,
 		Destination: destination,
 		User:        userName,
-	}))
-	if hErr != nil {
-		conn.Close()
-		n.NewError(ctx, E.Cause(hErr, "process connection from ", source))
+	})
+	if !waitForClose {
+		n.router.RouteConnectionEx(ctx, conn, metadata, nil)
+	} else {
+		done := make(chan struct{})
+		wrapper := v2rayhttp.NewHTTP2Wrapper(conn)
+		n.router.RouteConnectionEx(ctx, conn, metadata, N.OnceClose(func(it error) {
+			close(done)
+		}))
+		<-done
+		wrapper.CloseWrapper()
 	}
 }
 
 func (n *Naive) badRequest(ctx context.Context, request *http.Request, err error) {
-	n.NewError(ctx, E.Cause(err, "process connection from ", request.RemoteAddr))
+	n.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", request.RemoteAddr))
 }
 
 func rejectHTTP(writer http.ResponseWriter, statusCode int) {

inbound/redirect.go

@@ -9,7 +9,6 @@ import (
 	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
-	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
 )
@@ -34,11 +33,13 @@ func NewRedirect(ctx context.Context, router adapter.Router, logger log.ContextL
 	return redirect
 }
 
-func (r *Redirect) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
+func (r *Redirect) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
 	destination, err := redir.GetOriginalDestination(conn)
 	if err != nil {
-		return E.Cause(err, "get redirect destination")
+		conn.Close()
+		r.logger.ErrorContext(ctx, "process connection from ", conn.RemoteAddr(), ": get redirect destination: ", err)
+		return
 	}
 	metadata.Destination = M.SocksaddrFromNetIP(destination)
-	return r.newConnection(ctx, conn, metadata)
+	r.newConnectionEx(ctx, conn, metadata, onClose)
 }

inbound/shadowsocks.go

@@ -3,7 +3,6 @@ package inbound
 import (
 	"context"
 	"net"
-	"os"
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
@@ -18,6 +17,7 @@ import (
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/buf"
 	E "github.com/sagernet/sing/common/exceptions"
+	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/sing/common/ntp"
 )
@@ -36,8 +36,8 @@ func NewShadowsocks(ctx context.Context, router adapter.Router, logger log.Conte
 }
 
 var (
-	_ adapter.Inbound           = (*Shadowsocks)(nil)
-	_ adapter.InjectableInbound = (*Shadowsocks)(nil)
+	_ adapter.Inbound              = (*Shadowsocks)(nil)
+	_ adapter.TCPInjectableInbound = (*Shadowsocks)(nil)
 )
 
 type Shadowsocks struct {
@@ -74,11 +74,11 @@ func newShadowsocks(ctx context.Context, router adapter.Router, logger log.Conte
 	}
 	switch {
 	case options.Method == shadowsocks.MethodNone:
-		inbound.service = shadowsocks.NewNoneService(int64(udpTimeout.Seconds()), inbound.upstreamContextHandler())
+		inbound.service = shadowsocks.NewNoneService(int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound))
 	case common.Contains(shadowaead.List, options.Method):
-		inbound.service, err = shadowaead.NewService(options.Method, nil, options.Password, int64(udpTimeout.Seconds()), inbound.upstreamContextHandler())
+		inbound.service, err = shadowaead.NewService(options.Method, nil, options.Password, int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound))
 	case common.Contains(shadowaead_2022.List, options.Method):
-		inbound.service, err = shadowaead_2022.NewServiceWithPassword(options.Method, options.Password, int64(udpTimeout.Seconds()), inbound.upstreamContextHandler(), ntp.TimeFuncFromContext(ctx))
+		inbound.service, err = shadowaead_2022.NewServiceWithPassword(options.Method, options.Password, int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), ntp.TimeFuncFromContext(ctx))
 	default:
 		err = E.New("unsupported method: ", options.Method)
 	}
@@ -86,14 +86,29 @@ func newShadowsocks(ctx context.Context, router adapter.Router, logger log.Conte
 	return inbound, err
 }
 
-func (h *Shadowsocks) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	return h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, adapter.UpstreamMetadata(metadata))
+func (h *Shadowsocks) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
-func (h *Shadowsocks) NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, metadata adapter.InboundContext) error {
-	return h.service.NewPacket(adapter.WithContext(ctx, &metadata), conn, buffer, adapter.UpstreamMetadata(metadata))
+func (h *Shadowsocks) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) {
+	err := h.service.NewPacket(h.ctx, h.packetConn(), buffer, M.Metadata{Source: source})
+	if err != nil {
+		h.logger.Error(E.Cause(err, "process packet from ", source))
+	}
 }
 
-func (h *Shadowsocks) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *Shadowsocks) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
+	h.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
+	return h.router.RouteConnection(ctx, conn, h.createMetadata(conn, metadata))
+}
+
+func (h *Shadowsocks) newPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
+	ctx = log.ContextWithNewID(ctx)
+	h.logger.InfoContext(ctx, "inbound packet connection from ", metadata.Source)
+	h.logger.InfoContext(ctx, "inbound packet connection to ", metadata.Destination)
+	return h.router.RoutePacketConnection(ctx, conn, h.createPacketMetadata(conn, metadata))
 }

inbound/shadowsocks_multi.go

@@ -20,13 +20,14 @@ import (
 	"github.com/sagernet/sing/common/buf"
 	E "github.com/sagernet/sing/common/exceptions"
 	F "github.com/sagernet/sing/common/format"
+	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/sing/common/ntp"
 )
 
 var (
-	_ adapter.Inbound           = (*ShadowsocksMulti)(nil)
-	_ adapter.InjectableInbound = (*ShadowsocksMulti)(nil)
+	_ adapter.Inbound              = (*ShadowsocksMulti)(nil)
+	_ adapter.TCPInjectableInbound = (*ShadowsocksMulti)(nil)
 )
 
 type ShadowsocksMulti struct {
@@ -66,14 +67,15 @@ func newShadowsocksMulti(ctx context.Context, router adapter.Router, logger log.
 			options.Method,
 			options.Password,
 			int64(udpTimeout.Seconds()),
-			adapter.NewUpstreamContextHandler(inbound.newConnection, inbound.newPacketConnection, inbound),
+			adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
 			ntp.TimeFuncFromContext(ctx),
 		)
 	} else if common.Contains(shadowaead.List, options.Method) {
 		service, err = shadowaead.NewMultiService[int](
 			options.Method,
 			int64(udpTimeout.Seconds()),
-			adapter.NewUpstreamContextHandler(inbound.newConnection, inbound.newPacketConnection, inbound))
+			adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
+		)
 	} else {
 		return nil, E.New("unsupported method: " + options.Method)
 	}
@@ -94,16 +96,19 @@ func newShadowsocksMulti(ctx context.Context, router adapter.Router, logger log.
 	return inbound, err
 }
 
-func (h *ShadowsocksMulti) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	return h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, adapter.UpstreamMetadata(metadata))
+func (h *ShadowsocksMulti) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
-func (h *ShadowsocksMulti) NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, metadata adapter.InboundContext) error {
-	return h.service.NewPacket(adapter.WithContext(ctx, &metadata), conn, buffer, adapter.UpstreamMetadata(metadata))
-}
-
-func (h *ShadowsocksMulti) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *ShadowsocksMulti) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) {
+	err := h.service.NewPacket(h.ctx, h.packetConn(), buffer, M.Metadata{Source: source})
+	if err != nil {
+		h.logger.Error(E.Cause(err, "process packet from ", source))
+	}
 }
 
 func (h *ShadowsocksMulti) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
@@ -118,7 +123,7 @@ func (h *ShadowsocksMulti) newConnection(ctx context.Context, conn net.Conn, met
 		metadata.User = user
 	}
 	h.logger.InfoContext(ctx, "[", user, "] inbound connection to ", metadata.Destination)
-	return h.router.RouteConnection(ctx, conn, metadata)
+	return h.router.RouteConnection(ctx, conn, h.createMetadata(conn, metadata))
 }
 
 func (h *ShadowsocksMulti) newPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
@@ -135,5 +140,5 @@ func (h *ShadowsocksMulti) newPacketConnection(ctx context.Context, conn N.Packe
 	ctx = log.ContextWithNewID(ctx)
 	h.logger.InfoContext(ctx, "[", user, "] inbound packet connection from ", metadata.Source)
 	h.logger.InfoContext(ctx, "[", user, "] inbound packet connection to ", metadata.Destination)
-	return h.router.RoutePacketConnection(ctx, conn, metadata)
+	return h.router.RoutePacketConnection(ctx, conn, h.createPacketMetadata(conn, metadata))
 }

inbound/shadowsocks_relay.go

@@ -16,13 +16,15 @@ import (
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/auth"
 	"github.com/sagernet/sing/common/buf"
+	E "github.com/sagernet/sing/common/exceptions"
 	F "github.com/sagernet/sing/common/format"
+	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
 )
 
 var (
-	_ adapter.Inbound           = (*ShadowsocksRelay)(nil)
-	_ adapter.InjectableInbound = (*ShadowsocksRelay)(nil)
+	_ adapter.Inbound              = (*ShadowsocksRelay)(nil)
+	_ adapter.TCPInjectableInbound = (*ShadowsocksRelay)(nil)
 )
 
 type ShadowsocksRelay struct {
@@ -61,7 +63,7 @@ func newShadowsocksRelay(ctx context.Context, router adapter.Router, logger log.
 		options.Method,
 		options.Password,
 		int64(udpTimeout.Seconds()),
-		adapter.NewUpstreamContextHandler(inbound.newConnection, inbound.newPacketConnection, inbound),
+		adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
 	)
 	if err != nil {
 		return nil, err
@@ -79,16 +81,19 @@ func newShadowsocksRelay(ctx context.Context, router adapter.Router, logger log.
 	return inbound, err
 }
 
-func (h *ShadowsocksRelay) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	return h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, adapter.UpstreamMetadata(metadata))
+func (h *ShadowsocksRelay) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
-func (h *ShadowsocksRelay) NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, metadata adapter.InboundContext) error {
-	return h.service.NewPacket(adapter.WithContext(ctx, &metadata), conn, buffer, adapter.UpstreamMetadata(metadata))
-}
-
-func (h *ShadowsocksRelay) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *ShadowsocksRelay) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) {
+	err := h.service.NewPacket(h.ctx, h.packetConn(), buffer, M.Metadata{Source: source})
+	if err != nil {
+		h.logger.Error(E.Cause(err, "process packet from ", source))
+	}
 }
 
 func (h *ShadowsocksRelay) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
@@ -103,7 +108,7 @@ func (h *ShadowsocksRelay) newConnection(ctx context.Context, conn net.Conn, met
 		metadata.User = destination
 	}
 	h.logger.InfoContext(ctx, "[", destination, "] inbound connection to ", metadata.Destination)
-	return h.router.RouteConnection(ctx, conn, metadata)
+	return h.router.RouteConnection(ctx, conn, h.createMetadata(conn, metadata))
 }
 
 func (h *ShadowsocksRelay) newPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
@@ -120,5 +125,5 @@ func (h *ShadowsocksRelay) newPacketConnection(ctx context.Context, conn N.Packe
 	ctx = log.ContextWithNewID(ctx)
 	h.logger.InfoContext(ctx, "[", destination, "] inbound packet connection from ", metadata.Source)
 	h.logger.InfoContext(ctx, "[", destination, "] inbound packet connection to ", metadata.Destination)
-	return h.router.RoutePacketConnection(ctx, conn, metadata)
+	return h.router.RoutePacketConnection(ctx, conn, h.createPacketMetadata(conn, metadata))
 }

inbound/shadowtls.go

@@ -12,6 +12,7 @@ import (
 	"github.com/sagernet/sing-shadowtls"
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/auth"
+	E "github.com/sagernet/sing/common/exceptions"
 	N "github.com/sagernet/sing/common/network"
 )
 
@@ -91,3 +92,11 @@ func (h *ShadowTLS) newConnection(ctx context.Context, conn net.Conn, metadata a
 	}
 	return h.router.RouteConnection(ctx, conn, metadata)
 }
+
+func (h *ShadowTLS) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.NewConnection(ctx, conn, metadata)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
+}

inbound/socks.go

@@ -1,9 +1,9 @@
 package inbound
 
 import (
+	std_bufio "bufio"
 	"context"
 	"net"
-	"os"
 
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/uot"
@@ -11,13 +11,14 @@ import (
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
 	"github.com/sagernet/sing/common/auth"
+	E "github.com/sagernet/sing/common/exceptions"
 	N "github.com/sagernet/sing/common/network"
 	"github.com/sagernet/sing/protocol/socks"
 )
 
 var (
-	_ adapter.Inbound           = (*Socks)(nil)
-	_ adapter.InjectableInbound = (*Socks)(nil)
+	_ adapter.Inbound              = (*Socks)(nil)
+	_ adapter.TCPInjectableInbound = (*Socks)(nil)
 )
 
 type Socks struct {
@@ -42,10 +43,10 @@ func NewSocks(ctx context.Context, router adapter.Router, logger log.ContextLogg
 	return inbound
 }
 
-func (h *Socks) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	return socks.HandleConnection(ctx, conn, h.authenticator, h.upstreamUserHandler(metadata), adapter.UpstreamMetadata(metadata))
-}
-
-func (h *Socks) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *Socks) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := socks.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, nil, h.upstreamUserHandlerEx(metadata), metadata.Source, metadata.Destination, onClose)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }

inbound/tproxy.go

@@ -18,12 +18,12 @@ import (
 	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
-	"github.com/sagernet/sing/common/udpnat"
+	"github.com/sagernet/sing/common/udpnat2"
 )
 
 type TProxy struct {
 	myInboundAdapter
-	udpNat *udpnat.Service[netip.AddrPort]
+	udpNat *udpnat.Service
 }
 
 func NewTProxy(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.TProxyInboundOptions) *TProxy {
@@ -46,8 +46,7 @@ func NewTProxy(ctx context.Context, router adapter.Router, logger log.ContextLog
 	}
 	tproxy.connHandler = tproxy
 	tproxy.oobPacketHandler = tproxy
-	tproxy.udpNat = udpnat.New[netip.AddrPort](int64(udpTimeout.Seconds()), tproxy.upstreamContextHandler())
-	tproxy.packetUpstream = tproxy.udpNat
+	tproxy.udpNat = udpnat.New(tproxy, tproxy.preparePacketConnection, udpTimeout, false)
 	return tproxy
 }
 
@@ -75,35 +74,43 @@ func (t *TProxy) Start() error {
 	return nil
 }
 
-func (t *TProxy) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
+func (t *TProxy) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
 	metadata.Destination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap()
-	return t.newConnection(ctx, conn, metadata)
+	t.newConnectionEx(ctx, conn, metadata, onClose)
 }
 
-func (t *TProxy) NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, oob []byte, metadata adapter.InboundContext) error {
+func (t *TProxy) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	t.newPacketConnectionEx(ctx, conn, t.createPacketMetadataEx(source, destination), onClose)
+}
+
+func (t *TProxy) NewPacketEx(buffer *buf.Buffer, oob []byte, source M.Socksaddr) {
 	destination, err := redir.GetOriginalDestinationFromOOB(oob)
 	if err != nil {
-		return E.Cause(err, "get tproxy destination")
+		t.logger.Warn("process packet from ", source, ": get tproxy destination: ", err)
+		return
 	}
-	metadata.Destination = M.SocksaddrFromNetIP(destination).Unwrap()
-	t.udpNat.NewContextPacket(ctx, metadata.Source.AddrPort(), buffer, adapter.UpstreamMetadata(metadata), func(natConn N.PacketConn) (context.Context, N.PacketWriter) {
-		return adapter.WithContext(log.ContextWithNewID(ctx), &metadata), &tproxyPacketWriter{ctx: ctx, source: natConn, destination: metadata.Destination}
-	})
-	return nil
+	t.udpNat.NewPacket([][]byte{buffer.Bytes()}, source, M.SocksaddrFromNetIP(destination), nil)
 }
 
 type tproxyPacketWriter struct {
 	ctx         context.Context
-	source      N.PacketConn
+	source      netip.AddrPort
 	destination M.Socksaddr
 	conn        *net.UDPConn
 }
 
+func (t *TProxy) preparePacketConnection(source M.Socksaddr, destination M.Socksaddr, userData any) (bool, context.Context, N.PacketWriter, N.CloseHandlerFunc) {
+	writer := &tproxyPacketWriter{ctx: t.ctx, source: source.AddrPort(), destination: destination}
+	return true, t.ctx, writer, func(it error) {
+		common.Close(common.PtrOrNil(writer.conn))
+	}
+}
+
 func (w *tproxyPacketWriter) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
 	defer buffer.Release()
 	conn := w.conn
 	if w.destination == destination && conn != nil {
-		_, err := conn.WriteToUDPAddrPort(buffer.Bytes(), M.AddrPortFromNet(w.source.LocalAddr()))
+		_, err := conn.WriteToUDPAddrPort(buffer.Bytes(), w.source)
 		if err != nil {
 			w.conn = nil
 		}
@@ -122,9 +129,5 @@ func (w *tproxyPacketWriter) WritePacket(buffer *buf.Buffer, destination M.Socks
 	} else {
 		defer udpConn.Close()
 	}
-	return common.Error(udpConn.WriteToUDPAddrPort(buffer.Bytes(), M.AddrPortFromNet(w.source.LocalAddr())))
-}
-
-func (w *tproxyPacketWriter) Close() error {
-	return common.Close(common.PtrOrNil(w.conn))
+	return common.Error(udpConn.WriteToUDPAddrPort(buffer.Bytes(), w.source))
 }

inbound/trojan.go

@@ -22,8 +22,8 @@ import (
 )
 
 var (
-	_ adapter.Inbound           = (*Trojan)(nil)
-	_ adapter.InjectableInbound = (*Trojan)(nil)
+	_ adapter.Inbound              = (*Trojan)(nil)
+	_ adapter.TCPInjectableInbound = (*Trojan)(nil)
 )
 
 type Trojan struct {
@@ -90,7 +90,7 @@ func NewTrojan(ctx context.Context, router adapter.Router, logger log.ContextLog
 		return nil, err
 	}
 	if options.Transport != nil {
-		inbound.transport, err = v2ray.NewServerTransport(ctx, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*trojanTransportHandler)(inbound))
+		inbound.transport, err = v2ray.NewServerTransport(ctx, logger, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*trojanTransportHandler)(inbound))
 		if err != nil {
 			return nil, E.Cause(err, "create server transport: ", options.Transport.Type)
 		}
@@ -149,11 +149,6 @@ func (h *Trojan) Close() error {
 	)
 }
 
-func (h *Trojan) newTransportConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	h.injectTCP(conn, metadata)
-	return nil
-}
-
 func (h *Trojan) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
 	var err error
 	if h.tlsConfig != nil && h.transport == nil {
@@ -165,8 +160,12 @@ func (h *Trojan) NewConnection(ctx context.Context, conn net.Conn, metadata adap
 	return h.service.NewConnection(adapter.WithContext(ctx, &metadata), conn, adapter.UpstreamMetadata(metadata))
 }
 
-func (h *Trojan) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *Trojan) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.NewConnection(ctx, conn, metadata)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
 func (h *Trojan) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
@@ -226,9 +225,6 @@ var _ adapter.V2RayServerTransportHandler = (*trojanTransportHandler)(nil)
 
 type trojanTransportHandler Trojan
 
-func (t *trojanTransportHandler) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
-	return (*Trojan)(t).newTransportConnection(ctx, conn, adapter.InboundContext{
-		Source:      metadata.Source,
-		Destination: metadata.Destination,
-	})
+func (t *trojanTransportHandler) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	(*Trojan)(t).routeTCP(ctx, conn, source, destination, onClose)
 }

inbound/tun.go

@@ -28,17 +28,18 @@ import (
 	"go4.org/netipx"
 )
 
-var _ adapter.Inbound = (*Tun)(nil)
+var _ adapter.Inbound = (*TUN)(nil)
 
-type Tun struct {
-	tag                         string
-	ctx                         context.Context
-	router                      adapter.Router
-	logger                      log.ContextLogger
+type TUN struct {
+	tag    string
+	ctx    context.Context
+	router adapter.Router
+	logger log.ContextLogger
+	// Deprecated
 	inboundOptions              option.InboundOptions
 	tunOptions                  tun.Options
 	endpointIndependentNat      bool
-	udpTimeout                  int64
+	udpTimeout                  time.Duration
 	stack                       string
 	tunIf                       tun.Tun
 	tunStack                    tun.Stack
@@ -53,7 +54,7 @@ type Tun struct {
 	routeExcludeAddressSet      []*netipx.IPSet
 }
 
-func NewTun(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.TunInboundOptions, platformInterface platform.Interface) (*Tun, error) {
+func NewTun(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.TunInboundOptions, platformInterface platform.Interface) (*TUN, error) {
 	address := options.Address
 	var deprecatedAddressUsed bool
 	//nolint:staticcheck
@@ -162,7 +163,7 @@ func NewTun(ctx context.Context, router adapter.Router, logger log.ContextLogger
 		outputMark = tun.DefaultAutoRedirectOutputMark
 	}
 
-	inbound := &Tun{
+	inbound := &TUN{
 		tag:            tag,
 		ctx:            ctx,
 		router:         router,
@@ -194,7 +195,7 @@ func NewTun(ctx context.Context, router adapter.Router, logger log.ContextLogger
 			InterfaceMonitor:         router.InterfaceMonitor(),
 		},
 		endpointIndependentNat: options.EndpointIndependentNat,
-		udpTimeout:             int64(udpTimeout.Seconds()),
+		udpTimeout:             udpTimeout,
 		stack:                  options.Stack,
 		platformInterface:      platformInterface,
 		platformOptions:        common.PtrValueOrDefault(options.Platform),
@@ -207,7 +208,7 @@ func NewTun(ctx context.Context, router adapter.Router, logger log.ContextLogger
 		inbound.autoRedirect, err = tun.NewAutoRedirect(tun.AutoRedirectOptions{
 			TunOptions:             &inbound.tunOptions,
 			Context:                ctx,
-			Handler:                inbound,
+			Handler:                (*autoRedirectHandler)(inbound),
 			Logger:                 logger,
 			NetworkMonitor:         router.NetworkMonitor(),
 			InterfaceFinder:        router.InterfaceFinder(),
@@ -283,17 +284,17 @@ func parseRange(uidRanges []ranges.Range[uint32], rangeList []string) ([]ranges.
 	return uidRanges, nil
 }
 
-func (t *Tun) Type() string {
+func (t *TUN) Type() string {
 	return C.TypeTun
 }
 
-func (t *Tun) Tag() string {
+func (t *TUN) Tag() string {
 	return t.tag
 }
 
-func (t *Tun) Start() error {
+func (t *TUN) Start() error {
 	if C.IsAndroid && t.platformInterface == nil {
-		t.tunOptions.BuildAndroidRules(t.router.PackageManager(), t)
+		t.tunOptions.BuildAndroidRules(t.router.PackageManager())
 	}
 	if t.tunOptions.Name == "" {
 		t.tunOptions.Name = tun.CalculateInterfaceName("")
@@ -327,7 +328,6 @@ func (t *Tun) Start() error {
 		Context:                t.ctx,
 		Tun:                    tunInterface,
 		TunOptions:             t.tunOptions,
-		EndpointIndependentNat: t.endpointIndependentNat,
 		UDPTimeout:             t.udpTimeout,
 		Handler:                t,
 		Logger:                 t.logger,
@@ -349,7 +349,7 @@ func (t *Tun) Start() error {
 	return nil
 }
 
-func (t *Tun) PostStart() error {
+func (t *TUN) PostStart() error {
 	monitor := taskmonitor.New(t.logger, C.StartTimeout)
 	if t.autoRedirect != nil {
 		t.routeAddressSet = common.FlatMap(t.routeRuleSet, adapter.RuleSet.ExtractIPSet)
@@ -388,7 +388,7 @@ func (t *Tun) PostStart() error {
 	return nil
 }
 
-func (t *Tun) updateRouteAddressSet(it adapter.RuleSet) {
+func (t *TUN) updateRouteAddressSet(it adapter.RuleSet) {
 	t.routeAddressSet = common.FlatMap(t.routeRuleSet, adapter.RuleSet.ExtractIPSet)
 	t.routeExcludeAddressSet = common.FlatMap(t.routeExcludeRuleSet, adapter.RuleSet.ExtractIPSet)
 	t.autoRedirect.UpdateRouteAddressSet()
@@ -396,7 +396,7 @@ func (t *Tun) updateRouteAddressSet(it adapter.RuleSet) {
 	t.routeExcludeAddressSet = nil
 }
 
-func (t *Tun) Close() error {
+func (t *TUN) Close() error {
 	return common.Close(
 		t.tunStack,
 		t.tunIf,
@@ -404,44 +404,48 @@ func (t *Tun) Close() error {
 	)
 }
 
-func (t *Tun) NewConnection(ctx context.Context, conn net.Conn, upstreamMetadata M.Metadata) error {
-	ctx = log.ContextWithNewID(ctx)
-	var metadata adapter.InboundContext
-	metadata.Inbound = t.tag
-	metadata.InboundType = C.TypeTun
-	metadata.Source = upstreamMetadata.Source
-	metadata.Destination = upstreamMetadata.Destination
-	metadata.InboundOptions = t.inboundOptions
-	if upstreamMetadata.Protocol != "" {
-		t.logger.InfoContext(ctx, "inbound ", upstreamMetadata.Protocol, " connection from ", metadata.Source)
-	} else {
-		t.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
-	}
-	t.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
-	err := t.router.RouteConnection(ctx, conn, metadata)
-	if err != nil {
-		t.NewError(ctx, err)
-	}
+func (t *TUN) PrepareConnection(source M.Socksaddr, destination M.Socksaddr) error {
+	// TODO: implement rejects
 	return nil
 }
 
-func (t *Tun) NewPacketConnection(ctx context.Context, conn N.PacketConn, upstreamMetadata M.Metadata) error {
+func (t *TUN) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
 	ctx = log.ContextWithNewID(ctx)
 	var metadata adapter.InboundContext
 	metadata.Inbound = t.tag
 	metadata.InboundType = C.TypeTun
-	metadata.Source = upstreamMetadata.Source
-	metadata.Destination = upstreamMetadata.Destination
+	metadata.Source = source
+	metadata.Destination = destination
+	metadata.InboundOptions = t.inboundOptions
+	t.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
+	t.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
+	t.router.RouteConnectionEx(ctx, conn, metadata, onClose)
+}
+
+func (t *TUN) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	ctx = log.ContextWithNewID(ctx)
+	var metadata adapter.InboundContext
+	metadata.Inbound = t.tag
+	metadata.InboundType = C.TypeTun
+	metadata.Source = source
+	metadata.Destination = destination
 	metadata.InboundOptions = t.inboundOptions
 	t.logger.InfoContext(ctx, "inbound packet connection from ", metadata.Source)
 	t.logger.InfoContext(ctx, "inbound packet connection to ", metadata.Destination)
-	err := t.router.RoutePacketConnection(ctx, conn, metadata)
-	if err != nil {
-		t.NewError(ctx, err)
-	}
-	return nil
+	t.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
 }
 
-func (t *Tun) NewError(ctx context.Context, err error) {
-	NewError(t.logger, ctx, err)
+type autoRedirectHandler TUN
+
+func (t *autoRedirectHandler) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	ctx = log.ContextWithNewID(ctx)
+	var metadata adapter.InboundContext
+	metadata.Inbound = t.tag
+	metadata.InboundType = C.TypeTun
+	metadata.Source = source
+	metadata.Destination = destination
+	metadata.InboundOptions = t.inboundOptions
+	t.logger.InfoContext(ctx, "inbound redirect connection from ", metadata.Source)
+	t.logger.InfoContext(ctx, "inbound connection to ", metadata.Destination)
+	t.router.RouteConnectionEx(ctx, conn, metadata, onClose)
 }

inbound/vless.go

@@ -25,8 +25,8 @@ import (
 )
 
 var (
-	_ adapter.Inbound           = (*VLESS)(nil)
-	_ adapter.InjectableInbound = (*VLESS)(nil)
+	_ adapter.Inbound              = (*VLESS)(nil)
+	_ adapter.TCPInjectableInbound = (*VLESS)(nil)
 )
 
 type VLESS struct {
@@ -73,7 +73,7 @@ func NewVLESS(ctx context.Context, router adapter.Router, logger log.ContextLogg
 		}
 	}
 	if options.Transport != nil {
-		inbound.transport, err = v2ray.NewServerTransport(ctx, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*vlessTransportHandler)(inbound))
+		inbound.transport, err = v2ray.NewServerTransport(ctx, logger, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*vlessTransportHandler)(inbound))
 		if err != nil {
 			return nil, E.Cause(err, "create server transport: ", options.Transport.Type)
 		}
@@ -128,11 +128,6 @@ func (h *VLESS) Close() error {
 	)
 }
 
-func (h *VLESS) newTransportConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	h.injectTCP(conn, metadata)
-	return nil
-}
-
 func (h *VLESS) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
 	var err error
 	if h.tlsConfig != nil && h.transport == nil {
@@ -144,8 +139,12 @@ func (h *VLESS) NewConnection(ctx context.Context, conn net.Conn, metadata adapt
 	return h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, adapter.UpstreamMetadata(metadata))
 }
 
-func (h *VLESS) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *VLESS) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.NewConnection(ctx, conn, metadata)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
 func (h *VLESS) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
@@ -188,9 +187,6 @@ var _ adapter.V2RayServerTransportHandler = (*vlessTransportHandler)(nil)
 
 type vlessTransportHandler VLESS
 
-func (t *vlessTransportHandler) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
-	return (*VLESS)(t).newTransportConnection(ctx, conn, adapter.InboundContext{
-		Source:      metadata.Source,
-		Destination: metadata.Destination,
-	})
+func (t *vlessTransportHandler) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	t.routeTCP(ctx, conn, source, destination, onClose)
 }

inbound/vmess.go

@@ -25,8 +25,8 @@ import (
 )
 
 var (
-	_ adapter.Inbound           = (*VMess)(nil)
-	_ adapter.InjectableInbound = (*VMess)(nil)
+	_ adapter.Inbound              = (*VMess)(nil)
+	_ adapter.TCPInjectableInbound = (*VMess)(nil)
 )
 
 type VMess struct {
@@ -83,7 +83,7 @@ func NewVMess(ctx context.Context, router adapter.Router, logger log.ContextLogg
 		}
 	}
 	if options.Transport != nil {
-		inbound.transport, err = v2ray.NewServerTransport(ctx, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*vmessTransportHandler)(inbound))
+		inbound.transport, err = v2ray.NewServerTransport(ctx, logger, common.PtrValueOrDefault(options.Transport), inbound.tlsConfig, (*vmessTransportHandler)(inbound))
 		if err != nil {
 			return nil, E.Cause(err, "create server transport: ", options.Transport.Type)
 		}
@@ -142,11 +142,6 @@ func (h *VMess) Close() error {
 	)
 }
 
-func (h *VMess) newTransportConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
-	h.injectTCP(conn, metadata)
-	return nil
-}
-
 func (h *VMess) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
 	var err error
 	if h.tlsConfig != nil && h.transport == nil {
@@ -158,8 +153,12 @@ func (h *VMess) NewConnection(ctx context.Context, conn net.Conn, metadata adapt
 	return h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, adapter.UpstreamMetadata(metadata))
 }
 
-func (h *VMess) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
-	return os.ErrInvalid
+func (h *VMess) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
+	err := h.NewConnection(ctx, conn, metadata)
+	N.CloseOnHandshakeFailure(conn, onClose, err)
+	if err != nil {
+		h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+	}
 }
 
 func (h *VMess) newConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext) error {
@@ -202,9 +201,6 @@ var _ adapter.V2RayServerTransportHandler = (*vmessTransportHandler)(nil)
 
 type vmessTransportHandler VMess
 
-func (t *vmessTransportHandler) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
-	return (*VMess)(t).newTransportConnection(ctx, conn, adapter.InboundContext{
-		Source:      metadata.Source,
-		Destination: metadata.Destination,
-	})
+func (t *vmessTransportHandler) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
+	(*VMess)(t).routeTCP(ctx, conn, source, destination, onClose)
 }