Add multiplexer

common/debugio/log.go

@@ -0,0 +1,87 @@
+package debugio
+
+import (
+	"net"
+
+	"github.com/sagernet/sing-box/log"
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
+	M "github.com/sagernet/sing/common/metadata"
+	N "github.com/sagernet/sing/common/network"
+)
+
+type LogConn struct {
+	N.ExtendedConn
+	logger log.Logger
+	prefix string
+}
+
+func NewLogConn(conn net.Conn, logger log.Logger, prefix string) N.ExtendedConn {
+	return &LogConn{bufio.NewExtendedConn(conn), logger, prefix}
+}
+
+func (c *LogConn) Read(p []byte) (n int, err error) {
+	n, err = c.ExtendedConn.Read(p)
+	if n > 0 {
+		c.logger.Debug(c.prefix, " read ", buf.EncodeHexString(p[:n]))
+	}
+	return
+}
+
+func (c *LogConn) Write(p []byte) (n int, err error) {
+	c.logger.Debug(c.prefix, " write ", buf.EncodeHexString(p))
+	return c.ExtendedConn.Write(p)
+}
+
+func (c *LogConn) ReadBuffer(buffer *buf.Buffer) error {
+	err := c.ExtendedConn.ReadBuffer(buffer)
+	if err == nil {
+		c.logger.Debug(c.prefix, " read buffer ", buf.EncodeHexString(buffer.Bytes()))
+	}
+	return err
+}
+
+func (c *LogConn) WriteBuffer(buffer *buf.Buffer) error {
+	c.logger.Debug(c.prefix, " write buffer ", buf.EncodeHexString(buffer.Bytes()))
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *LogConn) Upstream() any {
+	return c.ExtendedConn
+}
+
+type LogPacketConn struct {
+	N.NetPacketConn
+	logger log.Logger
+	prefix string
+}
+
+func NewLogPacketConn(conn net.PacketConn, logger log.Logger, prefix string) N.NetPacketConn {
+	return &LogPacketConn{bufio.NewPacketConn(conn), logger, prefix}
+}
+
+func (c *LogPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
+	n, addr, err = c.NetPacketConn.ReadFrom(p)
+	if n > 0 {
+		c.logger.Debug(c.prefix, " read from ", addr, " ", buf.EncodeHexString(p[:n]))
+	}
+	return
+}
+
+func (c *LogPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	c.logger.Debug(c.prefix, " write to ", addr, " ", buf.EncodeHexString(p))
+	return c.NetPacketConn.WriteTo(p, addr)
+}
+
+func (c *LogPacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) {
+	destination, err = c.NetPacketConn.ReadPacket(buffer)
+	if err == nil {
+		c.logger.Debug(c.prefix, " read packet from ", destination, " ", buf.EncodeHexString(buffer.Bytes()))
+	}
+	return
+}
+
+func (c *LogPacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
+	c.logger.Debug(c.prefix, " write packet to ", destination, " ", buf.EncodeHexString(buffer.Bytes()))
+	return c.NetPacketConn.WritePacket(buffer, destination)
+}

common/debugio/race.go

@@ -0,0 +1,48 @@
+package debugio
+
+import (
+	"net"
+	"sync"
+
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
+	N "github.com/sagernet/sing/common/network"
+)
+
+type RaceConn struct {
+	N.ExtendedConn
+	readAccess  sync.Mutex
+	writeAccess sync.Mutex
+}
+
+func NewRaceConn(conn net.Conn) N.ExtendedConn {
+	return &RaceConn{ExtendedConn: bufio.NewExtendedConn(conn)}
+}
+
+func (c *RaceConn) Read(p []byte) (n int, err error) {
+	c.readAccess.Lock()
+	defer c.readAccess.Unlock()
+	return c.ExtendedConn.Read(p)
+}
+
+func (c *RaceConn) Write(p []byte) (n int, err error) {
+	c.writeAccess.Lock()
+	defer c.writeAccess.Unlock()
+	return c.ExtendedConn.Write(p)
+}
+
+func (c *RaceConn) ReadBuffer(buffer *buf.Buffer) error {
+	c.readAccess.Lock()
+	defer c.readAccess.Unlock()
+	return c.ExtendedConn.ReadBuffer(buffer)
+}
+
+func (c *RaceConn) WriteBuffer(buffer *buf.Buffer) error {
+	c.writeAccess.Lock()
+	defer c.writeAccess.Unlock()
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *RaceConn) Upstream() any {
+	return c.ExtendedConn
+}

common/dialer/default.go

@@ -12,6 +12,7 @@ import (
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/control"
 	M "github.com/sagernet/sing/common/metadata"
+	N "github.com/sagernet/sing/common/network"
 
 	"github.com/database64128/tfo-go"
 )
@@ -124,7 +125,7 @@ func (d *DefaultDialer) DialContext(ctx context.Context, network string, address
 }
 
 func (d *DefaultDialer) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
-	return d.ListenConfig.ListenPacket(ctx, C.NetworkUDP, "")
+	return d.ListenConfig.ListenPacket(ctx, N.NetworkUDP, "")
 }
 
 func (d *DefaultDialer) Upstream() any {

common/dialer/router.go

@@ -5,7 +5,6 @@ import (
 	"net"
 
 	"github.com/sagernet/sing-box/adapter"
-	C "github.com/sagernet/sing-box/constant"
 	M "github.com/sagernet/sing/common/metadata"
 	N "github.com/sagernet/sing/common/network"
 )
@@ -23,7 +22,7 @@ func (d *RouterDialer) DialContext(ctx context.Context, network string, destinat
 }
 
 func (d *RouterDialer) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
-	return d.router.DefaultOutbound(C.NetworkUDP).ListenPacket(ctx, destination)
+	return d.router.DefaultOutbound(N.NetworkUDP).ListenPacket(ctx, destination)
 }
 
 func (d *RouterDialer) Upstream() any {

common/dialer/tls.go

@@ -112,7 +112,7 @@ func NewTLS(dialer N.Dialer, serverAddress string, options option.OutboundTLSOpt
 }
 
 func (d *TLSDialer) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
-	if network != C.NetworkTCP {
+	if network != N.NetworkTCP {
 		return nil, os.ErrInvalid
 	}
 	conn, err := d.dialer.DialContext(ctx, network, destination)

common/mux/client.go

@@ -0,0 +1,476 @@
+package mux
+
+import (
+	"context"
+	"encoding/binary"
+	"io"
+	"net"
+	"sync"
+
+	"github.com/sagernet/sing-box/option"
+	"github.com/sagernet/sing/common"
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
+	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/x/list"
+
+	"github.com/hashicorp/yamux"
+)
+
+var _ N.Dialer = (*Client)(nil)
+
+type Client struct {
+	access         sync.Mutex
+	connections    list.List[*yamux.Session]
+	ctx            context.Context
+	dialer         N.Dialer
+	maxConnections int
+	minStreams     int
+	maxStreams     int
+}
+
+func NewClient(ctx context.Context, dialer N.Dialer, maxConnections int, minStreams int, maxStreams int) *Client {
+	return &Client{
+		ctx:            ctx,
+		dialer:         dialer,
+		maxConnections: maxConnections,
+		minStreams:     minStreams,
+		maxStreams:     maxStreams,
+	}
+}
+
+func NewClientWithOptions(ctx context.Context, dialer N.Dialer, options option.MultiplexOptions) N.Dialer {
+	if !options.Enabled {
+		return dialer
+	}
+	if options.MaxConnections == 0 && options.MaxStreams == 0 {
+		options.MinStreams = 8
+	}
+	return NewClient(ctx, dialer, options.MaxConnections, options.MinStreams, options.MaxStreams)
+}
+
+func (c *Client) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
+	switch N.NetworkName(network) {
+	case N.NetworkTCP:
+		stream, err := c.openStream()
+		if err != nil {
+			return nil, err
+		}
+		return &ClientConn{Conn: stream, destination: destination}, nil
+	case N.NetworkUDP:
+		stream, err := c.openStream()
+		if err != nil {
+			return nil, err
+		}
+		return bufio.NewUnbindPacketConn(&ClientPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: destination}), nil
+	default:
+		return nil, E.Extend(N.ErrUnknownNetwork, network)
+	}
+}
+
+func (c *Client) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
+	stream, err := c.openStream()
+	if err != nil {
+		return nil, err
+	}
+	// return bufio.NewUnbindPacketConn(&ClientPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: destination}), nil
+	return &ClientPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: destination}, nil
+}
+
+func (c *Client) openStream() (net.Conn, error) {
+	session, err := c.offer()
+	if err != nil {
+		return nil, err
+	}
+	conn, err := session.Open()
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+
+func (c *Client) offer() (*yamux.Session, error) {
+	c.access.Lock()
+	defer c.access.Unlock()
+
+	sessions := make([]*yamux.Session, 0, c.maxConnections)
+	for element := c.connections.Front(); element != nil; {
+		if element.Value.IsClosed() {
+			nextElement := element.Next()
+			c.connections.Remove(element)
+			element = nextElement
+			continue
+		}
+		sessions = append(sessions, element.Value)
+		element = element.Next()
+	}
+	sLen := len(sessions)
+	if sLen == 0 {
+		return c.offerNew()
+	}
+	// session := common.MinBy(sessions, yamux.Session.NumStreams)
+	session := common.MinBy(sessions, func(it *yamux.Session) int {
+		return it.NumStreams()
+	})
+	numStreams := session.NumStreams()
+	if numStreams == 0 {
+		return session, nil
+	}
+	if c.maxConnections > 0 {
+		if sLen >= c.maxConnections || numStreams < c.minStreams {
+			return session, nil
+		}
+	} else {
+		if c.maxStreams > 0 && numStreams < c.maxStreams {
+			return session, nil
+		}
+	}
+	return c.offerNew()
+}
+
+func (c *Client) offerNew() (*yamux.Session, error) {
+	conn, err := c.dialer.DialContext(c.ctx, N.NetworkTCP, Destination)
+	if err != nil {
+		return nil, err
+	}
+	session, err := yamux.Client(conn, newMuxConfig())
+	if err != nil {
+		return nil, err
+	}
+	c.connections.PushBack(session)
+	return session, nil
+}
+
+func (c *Client) Close() error {
+	c.access.Lock()
+	defer c.access.Unlock()
+	for _, session := range c.connections.Array() {
+		session.Close()
+	}
+	return nil
+}
+
+type ClientConn struct {
+	net.Conn
+	destination  M.Socksaddr
+	requestWrite bool
+	responseRead bool
+}
+
+func (c *ClientConn) readResponse() error {
+	response, err := ReadResponse(c.Conn)
+	if err != nil {
+		return err
+	}
+	if response.Status == statusError {
+		return E.New("remote error: ", response.Message)
+	}
+	return nil
+}
+
+func (c *ClientConn) Read(b []byte) (n int, err error) {
+	if !c.responseRead {
+		err = c.readResponse()
+		if err != nil {
+			return
+		}
+		c.responseRead = true
+	}
+	return c.Conn.Read(b)
+}
+
+func (c *ClientConn) Write(b []byte) (n int, err error) {
+	if c.requestWrite {
+		return c.Conn.Write(b)
+	}
+	request := Request{
+		Network:     N.NetworkTCP,
+		Destination: c.destination,
+	}
+	_buffer := buf.StackNewSize(requestLen(request) + len(b))
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	EncodeRequest(request, buffer)
+	buffer.Write(b)
+	_, err = c.Conn.Write(buffer.Bytes())
+	if err != nil {
+		return
+	}
+	c.requestWrite = true
+	return len(b), nil
+}
+
+func (c *ClientConn) ReadFrom(r io.Reader) (n int64, err error) {
+	if !c.requestWrite {
+		return bufio.ReadFrom0(c, r)
+	}
+	return bufio.Copy(c.Conn, r)
+}
+
+func (c *ClientConn) WriteTo(w io.Writer) (n int64, err error) {
+	if !c.responseRead {
+		return bufio.WriteTo0(c, w)
+	}
+	return bufio.Copy(w, c.Conn)
+}
+
+func (c *ClientConn) LocalAddr() net.Addr {
+	return c.Conn.LocalAddr()
+}
+
+func (c *ClientConn) RemoteAddr() net.Addr {
+	return c.destination.TCPAddr()
+}
+
+func (c *ClientConn) ReaderReplaceable() bool {
+	return c.responseRead
+}
+
+func (c *ClientConn) WriterReplaceable() bool {
+	return c.requestWrite
+}
+
+func (c *ClientConn) Upstream() any {
+	return c.Conn
+}
+
+type ClientPacketConn struct {
+	N.ExtendedConn
+	destination  M.Socksaddr
+	requestWrite bool
+	responseRead bool
+}
+
+func (c *ClientPacketConn) readResponse() error {
+	response, err := ReadResponse(c.ExtendedConn)
+	if err != nil {
+		return err
+	}
+	if response.Status == statusError {
+		return E.New("remote error: ", response.Message)
+	}
+	return nil
+}
+
+func (c *ClientPacketConn) Read(b []byte) (n int, err error) {
+	if !c.responseRead {
+		err = c.readResponse()
+		if err != nil {
+			return
+		}
+		c.responseRead = true
+	}
+	var length uint16
+	err = binary.Read(c.ExtendedConn, binary.BigEndian, &length)
+	if err != nil {
+		return
+	}
+	if cap(b) < int(length) {
+		return 0, io.ErrShortBuffer
+	}
+	return io.ReadFull(c.ExtendedConn, b[:length])
+}
+
+func (c *ClientPacketConn) writeRequest(payload []byte) (n int, err error) {
+	request := Request{
+		Network:     N.NetworkUDP,
+		Destination: c.destination,
+	}
+	rLen := requestLen(request)
+	if len(payload) > 0 {
+		rLen += 2 + len(payload)
+	}
+	_buffer := buf.StackNewSize(rLen)
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	EncodeRequest(request, buffer)
+	if len(payload) > 0 {
+		common.Must(
+			binary.Write(buffer, binary.BigEndian, uint16(len(payload))),
+			common.Error(buffer.Write(payload)),
+		)
+	}
+	_, err = c.ExtendedConn.Write(buffer.Bytes())
+	if err != nil {
+		return
+	}
+	c.requestWrite = true
+	return len(payload), nil
+}
+
+func (c *ClientPacketConn) Write(b []byte) (n int, err error) {
+	if !c.requestWrite {
+		return c.writeRequest(b)
+	}
+	err = binary.Write(c.ExtendedConn, binary.BigEndian, uint16(len(b)))
+	if err != nil {
+		return
+	}
+	return c.ExtendedConn.Write(b)
+}
+
+func (c *ClientPacketConn) WriteBuffer(buffer *buf.Buffer) error {
+	if !c.requestWrite {
+		defer buffer.Release()
+		return common.Error(c.writeRequest(buffer.Bytes()))
+	}
+	bLen := buffer.Len()
+	binary.BigEndian.PutUint16(buffer.ExtendHeader(2), uint16(bLen))
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *ClientPacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
+	return c.WriteBuffer(buffer)
+}
+
+func (c *ClientPacketConn) LocalAddr() net.Addr {
+	return c.ExtendedConn.LocalAddr()
+}
+
+func (c *ClientPacketConn) RemoteAddr() net.Addr {
+	return c.destination.UDPAddr()
+}
+
+func (c *ClientPacketConn) Upstream() any {
+	return c.ExtendedConn
+}
+
+var _ N.NetPacketConn = (*ClientPacketAddrConn)(nil)
+
+type ClientPacketAddrConn struct {
+	N.ExtendedConn
+	destination  M.Socksaddr
+	requestWrite bool
+	responseRead bool
+}
+
+func (c *ClientPacketAddrConn) readResponse() error {
+	response, err := ReadResponse(c.ExtendedConn)
+	if err != nil {
+		return err
+	}
+	if response.Status == statusError {
+		return E.New("remote error: ", response.Message)
+	}
+	return nil
+}
+
+func (c *ClientPacketAddrConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
+	if !c.responseRead {
+		err = c.readResponse()
+		if err != nil {
+			return
+		}
+		c.responseRead = true
+	}
+	destination, err := M.SocksaddrSerializer.ReadAddrPort(c.ExtendedConn)
+	if err != nil {
+		return
+	}
+	addr = destination.UDPAddr()
+	var length uint16
+	err = binary.Read(c.ExtendedConn, binary.BigEndian, &length)
+	if err != nil {
+		return
+	}
+	if cap(p) < int(length) {
+		return 0, nil, io.ErrShortBuffer
+	}
+	n, err = io.ReadFull(c.ExtendedConn, p[:length])
+	return
+}
+
+func (c *ClientPacketAddrConn) writeRequest(payload []byte, destination M.Socksaddr) (n int, err error) {
+	request := Request{
+		Network:     N.NetworkUDP,
+		Destination: c.destination,
+		PacketAddr:  true,
+	}
+	rLen := requestLen(request)
+	if len(payload) > 0 {
+		rLen += M.SocksaddrSerializer.AddrPortLen(destination) + 2 + len(payload)
+	}
+	_buffer := buf.StackNewSize(rLen)
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	EncodeRequest(request, buffer)
+	if len(payload) > 0 {
+		common.Must(
+			M.SocksaddrSerializer.WriteAddrPort(buffer, destination),
+			binary.Write(buffer, binary.BigEndian, uint16(len(payload))),
+			common.Error(buffer.Write(payload)),
+		)
+	}
+	_, err = c.ExtendedConn.Write(buffer.Bytes())
+	if err != nil {
+		return
+	}
+	c.requestWrite = true
+	return len(payload), nil
+}
+
+func (c *ClientPacketAddrConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	if !c.requestWrite {
+		return c.writeRequest(p, M.SocksaddrFromNet(addr))
+	}
+	err = M.SocksaddrSerializer.WriteAddrPort(c.ExtendedConn, M.SocksaddrFromNet(addr))
+	if err != nil {
+		return
+	}
+	err = binary.Write(c.ExtendedConn, binary.BigEndian, uint16(len(p)))
+	if err != nil {
+		return
+	}
+	return c.ExtendedConn.Write(p)
+}
+
+func (c *ClientPacketAddrConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) {
+	if !c.responseRead {
+		err = c.readResponse()
+		if err != nil {
+			return
+		}
+		c.responseRead = true
+	}
+	destination, err = M.SocksaddrSerializer.ReadAddrPort(c.ExtendedConn)
+	if err != nil {
+		return
+	}
+	var length uint16
+	err = binary.Read(c.ExtendedConn, binary.BigEndian, &length)
+	if err != nil {
+		return
+	}
+	if buffer.FreeLen() < int(length) {
+		return destination, io.ErrShortBuffer
+	}
+	_, err = io.ReadFull(c.ExtendedConn, buffer.Extend(int(length)))
+	return
+}
+
+func (c *ClientPacketAddrConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
+	if !c.requestWrite {
+		defer buffer.Release()
+		return common.Error(c.writeRequest(buffer.Bytes(), destination))
+	}
+	bLen := buffer.Len()
+	header := buf.With(buffer.ExtendHeader(M.SocksaddrSerializer.AddrPortLen(destination) + 2))
+	common.Must(
+		M.SocksaddrSerializer.WriteAddrPort(header, destination),
+		binary.Write(header, binary.BigEndian, uint16(bLen)),
+	)
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *ClientPacketAddrConn) LocalAddr() net.Addr {
+	return c.ExtendedConn.LocalAddr()
+}
+
+func (c *ClientPacketAddrConn) Upstream() any {
+	return c.ExtendedConn
+}

common/mux/protocol.go

@@ -0,0 +1,119 @@
+package mux
+
+import (
+	"encoding/binary"
+	"io"
+
+	C "github.com/sagernet/sing-box/constant"
+	"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/rw"
+
+	"github.com/hashicorp/yamux"
+)
+
+var Destination = M.Socksaddr{
+	Fqdn: "sp.mux.sing-box.arpa",
+	Port: 444,
+}
+
+func newMuxConfig() *yamux.Config {
+	config := yamux.DefaultConfig()
+	config.LogOutput = io.Discard
+	config.StreamCloseTimeout = C.TCPTimeout
+	config.StreamOpenTimeout = C.TCPTimeout
+	return config
+}
+
+const (
+	version0      = 0
+	flagUDP       = 1
+	flagAddr      = 2
+	statusSuccess = 0
+	statusError   = 1
+)
+
+type Request struct {
+	Network     string
+	Destination M.Socksaddr
+	PacketAddr  bool
+}
+
+func ReadRequest(reader io.Reader) (*Request, error) {
+	version, err := rw.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+	if version != version0 {
+		return nil, E.New("unsupported version: ", version)
+	}
+	var flags uint16
+	err = binary.Read(reader, binary.BigEndian, &flags)
+	if err != nil {
+		return nil, err
+	}
+	destination, err := M.SocksaddrSerializer.ReadAddrPort(reader)
+	if err != nil {
+		return nil, err
+	}
+	var network string
+	var udpAddr bool
+	if flags&flagUDP == 0 {
+		network = N.NetworkTCP
+	} else {
+		network = N.NetworkUDP
+		udpAddr = flags&flagAddr != 0
+	}
+	return &Request{network, destination, udpAddr}, nil
+}
+
+func requestLen(request Request) int {
+	var rLen int
+	rLen += 1 // version
+	rLen += 2 // flags
+	rLen += M.SocksaddrSerializer.AddrPortLen(request.Destination)
+	return rLen
+}
+
+func EncodeRequest(request Request, buffer *buf.Buffer) {
+	destination := request.Destination
+	var flags uint16
+	if request.Network == N.NetworkUDP {
+		flags |= flagUDP
+	}
+	if request.PacketAddr {
+		flags |= flagAddr
+		if !destination.IsValid() {
+			destination = Destination
+		}
+	}
+	common.Must(
+		buffer.WriteByte(version0),
+		binary.Write(buffer, binary.BigEndian, flags),
+		M.SocksaddrSerializer.WriteAddrPort(buffer, destination),
+	)
+}
+
+type Response struct {
+	Status  uint8
+	Message string
+}
+
+func ReadResponse(reader io.Reader) (*Response, error) {
+	var response Response
+	status, err := rw.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+	response.Status = status
+	if status == statusError {
+		response.Message, err = rw.ReadVString(reader)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return &response, nil
+}

common/mux/service.go

@@ -0,0 +1,210 @@
+package mux
+
+import (
+	"context"
+	"encoding/binary"
+	"net"
+
+	"github.com/sagernet/sing-box/adapter"
+	"github.com/sagernet/sing-box/log"
+	"github.com/sagernet/sing/common"
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
+	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/rw"
+
+	"github.com/hashicorp/yamux"
+)
+
+func NewConnection(ctx context.Context, router adapter.Router, errorHandler E.Handler, logger log.ContextLogger, conn net.Conn, metadata adapter.InboundContext) error {
+	session, err := yamux.Server(conn, newMuxConfig())
+	if err != nil {
+		return err
+	}
+	for {
+		stream, err := session.Accept()
+		if err != nil {
+			return err
+		}
+		request, err := ReadRequest(stream)
+		if err != nil {
+			return err
+		}
+		metadata.Destination = request.Destination
+		if request.Network == N.NetworkTCP {
+			go func() {
+				logger.InfoContext(ctx, "inbound multiplex connection to ", metadata.Destination)
+				hErr := router.RouteConnection(ctx, &ServerConn{ExtendedConn: bufio.NewExtendedConn(stream)}, metadata)
+				// hErr := router.RouteConnection(ctx, &ServerConn{ExtendedConn: bufio.NewExtendedConn(stream)}, metadata)
+				if hErr != nil {
+					errorHandler.NewError(ctx, hErr)
+				}
+			}()
+		} else {
+			go func() {
+				var packetConn N.PacketConn
+				if !request.PacketAddr {
+					logger.InfoContext(ctx, "inbound multiplex packet connection to ", metadata.Destination)
+					packetConn = &ServerPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: request.Destination}
+				} else {
+					logger.InfoContext(ctx, "inbound multiplex packet connection")
+					packetConn = &ServerPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream)}
+				}
+				hErr := router.RoutePacketConnection(ctx, packetConn, metadata)
+				if hErr != nil {
+					errorHandler.NewError(ctx, hErr)
+				}
+			}()
+		}
+	}
+}
+
+var _ N.HandshakeConn = (*ServerConn)(nil)
+
+type ServerConn struct {
+	N.ExtendedConn
+	responseWrite bool
+}
+
+func (c *ServerConn) HandshakeFailure(err error) error {
+	errMessage := err.Error()
+	_buffer := buf.StackNewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	common.Must(
+		buffer.WriteByte(statusError),
+		rw.WriteVString(_buffer, errMessage),
+	)
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *ServerConn) Write(b []byte) (n int, err error) {
+	if c.responseWrite {
+		return c.ExtendedConn.Write(b)
+	}
+	_buffer := buf.StackNewSize(1 + len(b))
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	common.Must(
+		buffer.WriteByte(statusSuccess),
+		common.Error(buffer.Write(b)),
+	)
+	_, err = c.ExtendedConn.Write(buffer.Bytes())
+	if err != nil {
+		return
+	}
+	c.responseWrite = true
+	return len(b), nil
+}
+
+func (c *ServerConn) WriteBuffer(buffer *buf.Buffer) error {
+	if c.responseWrite {
+		return c.ExtendedConn.WriteBuffer(buffer)
+	}
+	buffer.ExtendHeader(1)[0] = statusSuccess
+	c.responseWrite = true
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+var (
+	_ N.HandshakeConn = (*ServerPacketConn)(nil)
+	_ N.PacketConn    = (*ServerPacketConn)(nil)
+)
+
+type ServerPacketConn struct {
+	N.ExtendedConn
+	destination   M.Socksaddr
+	responseWrite bool
+}
+
+func (c *ServerPacketConn) HandshakeFailure(err error) error {
+	errMessage := err.Error()
+	_buffer := buf.StackNewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	common.Must(
+		buffer.WriteByte(statusError),
+		rw.WriteVString(_buffer, errMessage),
+	)
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *ServerPacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) {
+	var length uint16
+	err = binary.Read(c.ExtendedConn, binary.BigEndian, &length)
+	if err != nil {
+		return
+	}
+	_, err = buffer.ReadFullFrom(c.ExtendedConn, int(length))
+	if err != nil {
+		return
+	}
+	destination = c.destination
+	return
+}
+
+func (c *ServerPacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
+	pLen := buffer.Len()
+	common.Must(binary.Write(buf.With(buffer.ExtendHeader(2)), binary.BigEndian, uint16(pLen)))
+	if !c.responseWrite {
+		buffer.ExtendHeader(1)[0] = statusSuccess
+		c.responseWrite = true
+	}
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+var (
+	_ N.HandshakeConn = (*ServerPacketAddrConn)(nil)
+	_ N.PacketConn    = (*ServerPacketAddrConn)(nil)
+)
+
+type ServerPacketAddrConn struct {
+	N.ExtendedConn
+	responseWrite bool
+}
+
+func (c *ServerPacketAddrConn) HandshakeFailure(err error) error {
+	errMessage := err.Error()
+	_buffer := buf.StackNewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
+	defer common.KeepAlive(_buffer)
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	common.Must(
+		buffer.WriteByte(statusError),
+		rw.WriteVString(_buffer, errMessage),
+	)
+	return c.ExtendedConn.WriteBuffer(buffer)
+}
+
+func (c *ServerPacketAddrConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) {
+	destination, err = M.SocksaddrSerializer.ReadAddrPort(c.ExtendedConn)
+	if err != nil {
+		return
+	}
+	var length uint16
+	err = binary.Read(c.ExtendedConn, binary.BigEndian, &length)
+	if err != nil {
+		return
+	}
+	_, err = buffer.ReadFullFrom(c.ExtendedConn, int(length))
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (c *ServerPacketAddrConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
+	pLen := buffer.Len()
+	common.Must(binary.Write(buf.With(buffer.ExtendHeader(2)), binary.BigEndian, uint16(pLen)))
+	common.Must(M.SocksaddrSerializer.WriteAddrPort(buf.With(buffer.ExtendHeader(M.SocksaddrSerializer.AddrPortLen(destination))), destination))
+	if !c.responseWrite {
+		buffer.ExtendHeader(1)[0] = statusSuccess
+		c.responseWrite = true
+	}
+	return c.ExtendedConn.WriteBuffer(buffer)
+}

common/process/searcher_darwin.go

@@ -8,8 +8,8 @@ import (
 	"syscall"
 	"unsafe"
 
-	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/log"
+	N "github.com/sagernet/sing/common/network"
 
 	"golang.org/x/sys/unix"
 )
@@ -33,9 +33,9 @@ func (d *darwinSearcher) FindProcessInfo(ctx context.Context, network string, sr
 func findProcessName(network string, ip netip.Addr, port int) (string, error) {
 	var spath string
 	switch network {
-	case C.NetworkTCP:
+	case N.NetworkTCP:
 		spath = "net.inet.tcp.pcblist_n"
-	case C.NetworkUDP:
+	case N.NetworkUDP:
 		spath = "net.inet.udp.pcblist_n"
 	default:
 		return "", os.ErrInvalid
@@ -55,7 +55,7 @@ func findProcessName(network string, ip netip.Addr, port int) (string, error) {
 	// rup8(sizeof(xinpcb_n)) + rup8(sizeof(xsocket_n)) +
 	// 2 * rup8(sizeof(xsockbuf_n)) + rup8(sizeof(xsockstat_n))
 	itemSize := 384
-	if network == C.NetworkTCP {
+	if network == N.NetworkTCP {
 		// rup8(sizeof(xtcpcb_n))
 		itemSize += 208
 	}

common/process/searcher_linux_shared.go

@@ -15,10 +15,10 @@ import (
 	"unicode"
 	"unsafe"
 
-	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/buf"
 	E "github.com/sagernet/sing/common/exceptions"
+	N "github.com/sagernet/sing/common/network"
 )
 
 // from https://github.com/vishvananda/netlink/blob/bca67dfc8220b44ef582c9da4e9172bf1c9ec973/nl/nl_linux.go#L52-L62
@@ -52,9 +52,9 @@ func resolveSocketByNetlink0(network string, ip netip.Addr, srcPort int) (inode
 	var protocol byte
 
 	switch network {
-	case C.NetworkTCP:
+	case N.NetworkTCP:
 		protocol = syscall.IPPROTO_TCP
-	case C.NetworkUDP:
+	case N.NetworkUDP:
 		protocol = syscall.IPPROTO_UDP
 	default:
 		return 0, 0, os.ErrInvalid

common/process/searcher_windows.go

@@ -8,9 +8,9 @@ import (
 	"syscall"
 	"unsafe"
 
-	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/log"
 	E "github.com/sagernet/sing/common/exceptions"
+	N "github.com/sagernet/sing/common/network"
 
 	"golang.org/x/sys/windows"
 )
@@ -86,10 +86,10 @@ func findProcessName(network string, ip netip.Addr, srcPort int) (string, error)
 	var class int
 	var fn uintptr
 	switch network {
-	case C.NetworkTCP:
+	case N.NetworkTCP:
 		fn = procGetExtendedTcpTable.Addr()
 		class = tcpTablePidConn
-	case C.NetworkUDP:
+	case N.NetworkUDP:
 		fn = procGetExtendedUdpTable.Addr()
 		class = udpTablePid
 	default:
@@ -101,7 +101,7 @@ func findProcessName(network string, ip netip.Addr, srcPort int) (string, error)
 		return "", err
 	}
 
-	s := newSearcher(family == windows.AF_INET, network == C.NetworkTCP)
+	s := newSearcher(family == windows.AF_INET, network == N.NetworkTCP)
 
 	pid, err := s.Search(buf, ip, uint16(srcPort))
 	if err != nil {