diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 9de14e50..3fbd01af 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -46,21 +46,21 @@ jobs: goarch: amd64 steps: - name: Checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 with: fetch-depth: 0 - name: Setup Go - uses: actions/setup-go@v5 + uses: actions/setup-go@v7 with: go-version: ^1.25 - name: Cache go module - uses: actions/cache@v4 + uses: actions/cache@v6 with: path: | ~/go/pkg/mod key: go-${{ hashFiles('**/go.sum') }} - name: golangci-lint - uses: golangci/golangci-lint-action@v8 + uses: golangci/golangci-lint-action@v9 env: GOOS: ${{ matrix.goos }} GOARCH: ${{ matrix.goarch }} diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 704d7ff9..71c1e935 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -37,11 +37,11 @@ jobs: os-name: Windows steps: - name: Checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 with: fetch-depth: 0 - name: Setup Go - uses: actions/setup-go@v5 + uses: actions/setup-go@v7 with: go-version: ${{ matrix.go-version }} - name: Build diff --git a/common/bufio/bind.go b/common/bufio/bind.go index dd46fb83..6bbb81d6 100644 --- a/common/bufio/bind.go +++ b/common/bufio/bind.go @@ -2,6 +2,7 @@ package bufio import ( "net" + "sync" "github.com/sagernet/sing/common/buf" M "github.com/sagernet/sing/common/metadata" @@ -59,10 +60,8 @@ func (c *bindPacketConn) Upstream() any { } var ( - _ N.NetPacketConn = (*UnbindPacketConn)(nil) - _ N.PacketReadWaitCreator = (*UnbindPacketConn)(nil) - _ N.ConnectedPacketBatchReadWaitCreator = (*UnbindPacketConn)(nil) - _ N.ConnectedPacketBatchWriteCreator = (*UnbindPacketConn)(nil) + _ N.NetPacketConn = (*UnbindPacketConn)(nil) + _ N.PacketReadWaitCreator = (*UnbindPacketConn)(nil) ) type UnbindPacketConn struct { @@ -117,14 +116,6 @@ func (c *UnbindPacketConn) CreateReadWaiter() (N.PacketReadWaiter, bool) { return &unbindPacketReadWaiter{readWaiter, c.addr}, true } -func (c *UnbindPacketConn) CreateConnectedPacketBatchReadWaiter() (N.ConnectedPacketBatchReadWaiter, bool) { - return createSyscallConnectedPacketBatchReadWaiter(c.ExtendedConn, c.addr) -} - -func (c *UnbindPacketConn) CreateConnectedPacketBatchWriter() (N.ConnectedPacketBatchWriter, bool) { - return createSyscallConnectedPacketBatchWriter(c.ExtendedConn) -} - func (c *UnbindPacketConn) Upstream() any { return c.ExtendedConn } @@ -137,7 +128,8 @@ func NewServerPacketConn(conn net.PacketConn) N.ExtendedConn { type serverPacketConn struct { N.NetPacketConn - remoteAddr M.Socksaddr + remoteAccess sync.RWMutex + remoteAddr M.Socksaddr } func (c *serverPacketConn) Read(p []byte) (n int, err error) { @@ -145,7 +137,7 @@ func (c *serverPacketConn) Read(p []byte) (n int, err error) { if err != nil { return } - c.remoteAddr = M.SocksaddrFromNet(addr) + c.updateRemoteAddr(M.SocksaddrFromNet(addr).Unwrap()) return } @@ -154,20 +146,33 @@ func (c *serverPacketConn) ReadBuffer(buffer *buf.Buffer) error { if err != nil { return err } - c.remoteAddr = destination + c.updateRemoteAddr(destination) return nil } func (c *serverPacketConn) Write(p []byte) (n int, err error) { - return c.NetPacketConn.WriteTo(p, c.remoteAddr.UDPAddr()) + return c.NetPacketConn.WriteTo(p, c.remoteDestination().UDPAddr()) } func (c *serverPacketConn) WriteBuffer(buffer *buf.Buffer) error { - return c.NetPacketConn.WritePacket(buffer, c.remoteAddr) + return c.NetPacketConn.WritePacket(buffer, c.remoteDestination()) } func (c *serverPacketConn) RemoteAddr() net.Addr { - return c.remoteAddr + return c.remoteDestination() +} + +func (c *serverPacketConn) updateRemoteAddr(destination M.Socksaddr) { + c.remoteAccess.Lock() + c.remoteAddr = destination + c.remoteAccess.Unlock() +} + +func (c *serverPacketConn) remoteDestination() M.Socksaddr { + c.remoteAccess.RLock() + destination := c.remoteAddr + c.remoteAccess.RUnlock() + return destination } func (c *serverPacketConn) Upstream() any { diff --git a/common/bufio/bind_batch.go b/common/bufio/bind_batch.go new file mode 100644 index 00000000..4395fbe5 --- /dev/null +++ b/common/bufio/bind_batch.go @@ -0,0 +1,139 @@ +package bufio + +import ( + "sync" + + "github.com/sagernet/sing/common/buf" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" +) + +func (c *UnbindPacketConn) CreateConnectedPacketBatchWriter() (N.ConnectedPacketBatchWriter, bool) { + var packetWriter N.PacketWriter + var destination func() M.Socksaddr + upstream := N.UnwrapWriter(c.ExtendedConn) + switch conn := upstream.(type) { + case *bindPacketConn: + packetWriter = conn.NetPacketConn + address := M.SocksaddrFromNet(conn.addr).Unwrap() + destination = func() M.Socksaddr { return address } + case *serverPacketConn: + packetWriter = conn.NetPacketConn + destination = conn.remoteDestination + default: + return createSyscallConnectedPacketBatchWriter(c.ExtendedConn) + } + writer, created := CreatePacketBatchWriter(packetWriter) + if !created { + return nil, false + } + return &boundConnectedPacketBatchWriter{writer: writer, destination: destination}, true +} + +func (c *serverPacketConn) CreatePacketBatchWriter() (N.PacketBatchWriter, bool) { + return CreatePacketBatchWriter(c.NetPacketConn) +} + +type boundConnectedPacketBatchWriter struct { + writer N.PacketBatchWriter + destination func() M.Socksaddr + access sync.Mutex + destinations []M.Socksaddr +} + +func (w *boundConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers []*buf.Buffer) error { + count := 0 + for _, buffer := range buffers { + if buffer.IsEmpty() { + buffer.Release() + continue + } + buffers[count] = buffer + count++ + } + clear(buffers[count:]) + if count == 0 { + return nil + } + buffers = buffers[:count] + w.access.Lock() + defer w.access.Unlock() + if cap(w.destinations) < len(buffers) { + w.destinations = make([]M.Socksaddr, len(buffers)) + } else { + w.destinations = w.destinations[:len(buffers)] + } + destination := w.destination() + for index := range w.destinations { + w.destinations[index] = destination + } + return w.writer.WritePacketBatch(buffers, w.destinations) +} + +func (w *boundConnectedPacketBatchWriter) Upstream() any { return w.writer } + +func (c *UnbindPacketConn) CreateConnectedPacketBatchReadWaiter() (N.ConnectedPacketBatchReadWaiter, bool) { + upstream := N.UnwrapReader(c.ExtendedConn) + switch conn := upstream.(type) { + case *bindPacketConn: + reader, created := CreatePacketBatchReadWaiter(conn.NetPacketConn) + if !created { + return nil, false + } + return &bindConnectedPacketBatchReadWaiter{reader: reader, destination: c.addr}, true + case *serverPacketConn: + reader, created := CreatePacketBatchReadWaiter(conn.NetPacketConn) + if !created { + return nil, false + } + return &serverConnectedPacketBatchReadWaiter{conn: conn, reader: reader, destination: c.addr}, true + default: + return createSyscallConnectedPacketBatchReadWaiter(c.ExtendedConn, c.addr) + } +} + +type bindConnectedPacketBatchReadWaiter struct { + reader N.PacketBatchReadWaiter + destination M.Socksaddr +} + +func (r *bindConnectedPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) bool { + return r.reader.InitializeReadWaiter(options) +} + +func (r *bindConnectedPacketBatchReadWaiter) WaitReadConnectedPackets() ([]*buf.Buffer, M.Socksaddr, error) { + buffers, _, err := r.reader.WaitReadPackets() + return buffers, r.destination, err +} + +func (r *bindConnectedPacketBatchReadWaiter) Upstream() any { return r.reader } + +type serverConnectedPacketBatchReadWaiter struct { + conn *serverPacketConn + reader N.PacketBatchReadWaiter + destination M.Socksaddr +} + +func (r *serverConnectedPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) bool { + return r.reader.InitializeReadWaiter(options) +} + +func (r *serverConnectedPacketBatchReadWaiter) WaitReadConnectedPackets() ([]*buf.Buffer, M.Socksaddr, error) { + buffers, destinations, err := r.reader.WaitReadPackets() + if err == nil { + r.conn.updateRemoteAddr(destinations[len(destinations)-1]) + } + return buffers, r.destination, err +} + +func (r *serverConnectedPacketBatchReadWaiter) Upstream() any { return r.reader } + +var ( + _ N.PacketBatchReadWaitCreator = (*serverPacketConn)(nil) + _ N.ConnectedPacketBatchWriteCreator = (*UnbindPacketConn)(nil) + _ N.ConnectedPacketBatchReadWaitCreator = (*UnbindPacketConn)(nil) + _ N.PacketBatchWriteCreator = (*serverPacketConn)(nil) + _ N.ConnectedPacketBatchWriter = (*boundConnectedPacketBatchWriter)(nil) + _ N.ConnectedPacketBatchReadWaiter = (*bindConnectedPacketBatchReadWaiter)(nil) + _ N.ConnectedPacketBatchReadWaiter = (*serverConnectedPacketBatchReadWaiter)(nil) +) diff --git a/common/bufio/bind_wait.go b/common/bufio/bind_wait.go index 779474ca..ee33800d 100644 --- a/common/bufio/bind_wait.go +++ b/common/bufio/bind_wait.go @@ -57,6 +57,6 @@ func (w *serverPacketReadWaiter) WaitReadBuffer() (buffer *buf.Buffer, err error if err != nil { return } - w.remoteAddr = destination + w.updateRemoteAddr(destination) return } diff --git a/common/bufio/cache.go b/common/bufio/cache.go index 94423887..e8df7034 100644 --- a/common/bufio/cache.go +++ b/common/bufio/cache.go @@ -179,15 +179,12 @@ func NewCachedPacketConn(conn N.PacketConn, buffer *buf.Buffer, destination M.So } func (c *CachedPacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) { - if c.buffer != nil { - _, err = buffer.ReadOnceFrom(c.buffer) - if err != nil { - return M.Socksaddr{}, err - } - c.buffer.DecRef() - c.buffer.Release() - c.buffer = nil - return c.destination, nil + packet := c.ReadCachedPacket() + if packet != nil { + defer packet.Buffer.Release() + defer N.PutPacketBuffer(packet) + _, err = buffer.Write(packet.Buffer.Bytes()) + return packet.Destination, err } return c.PacketConn.ReadPacket(buffer) } @@ -198,9 +195,10 @@ func (c *CachedPacketConn) ReadCachedPacket() *N.PacketBuffer { } buffer := c.buffer c.buffer = nil - if buffer != nil { - buffer.DecRef() + if buffer == nil { + return nil } + buffer.DecRef() packet := N.NewPacketBuffer() *packet = N.PacketBuffer{ Buffer: buffer, @@ -214,7 +212,7 @@ func (c *CachedPacketConn) Upstream() any { } func (c *CachedPacketConn) ReaderReplaceable() bool { - return c.buffer == nil + return c.taken.Load() } func (c *CachedPacketConn) WriterReplaceable() bool { diff --git a/common/bufio/copy_direct.go b/common/bufio/copy_direct.go index a460c7ed..0263c22a 100644 --- a/common/bufio/copy_direct.go +++ b/common/bufio/copy_direct.go @@ -3,7 +3,6 @@ package bufio import ( "errors" "io" - "os" "github.com/sagernet/sing/common/buf" M "github.com/sagernet/sing/common/metadata" @@ -140,10 +139,6 @@ func copyPacketBatchWaitWithPool(session *packetCopySession, destinationConn N.P if err != nil { return handled, n, err } - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return handled, n, os.ErrInvalid - } dataLens := make([]int, len(buffers)) for index, buffer := range buffers { dataLens[index] = buffer.Len() @@ -171,18 +166,11 @@ func copyPacketBatchWaitWithPool(session *packetCopySession, destinationConn N.P func copyPacketBatchToConnectedWaitWithPool(session *packetCopySession, destinationConn N.ConnectedPacketBatchWriter, source N.PacketBatchReadWaiter, notFirstTime bool) (handled bool, n int64, err error) { handled = true for { - var ( - buffers []*buf.Buffer - destinations []M.Socksaddr - ) - buffers, destinations, err = source.WaitReadPackets() + var buffers []*buf.Buffer + buffers, _, err = source.WaitReadPackets() if err != nil { return handled, n, err } - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return handled, n, os.ErrInvalid - } dataLens := make([]int, len(buffers)) for index, buffer := range buffers { dataLens[index] = buffer.Len() @@ -218,10 +206,6 @@ func copyConnectedPacketBatchWaitWithPool(session *packetCopySession, destinatio if err != nil { return handled, n, err } - if len(buffers) == 0 { - buf.ReleaseMulti(buffers) - return handled, n, os.ErrInvalid - } destinations := make([]M.Socksaddr, len(buffers)) dataLens := make([]int, len(buffers)) for index, buffer := range buffers { @@ -256,10 +240,6 @@ func copyConnectedPacketBatchToConnectedWaitWithPool(session *packetCopySession, if err != nil { return handled, n, err } - if len(buffers) == 0 { - buf.ReleaseMulti(buffers) - return handled, n, os.ErrInvalid - } dataLens := make([]int, len(buffers)) for index, buffer := range buffers { dataLens[index] = buffer.Len() diff --git a/common/bufio/counter_packet_conn.go b/common/bufio/counter_packet_conn.go index 2208d1d0..b85f4dfd 100644 --- a/common/bufio/counter_packet_conn.go +++ b/common/bufio/counter_packet_conn.go @@ -1,7 +1,6 @@ package bufio import ( - "os" "sync/atomic" "github.com/sagernet/sing/common" @@ -84,10 +83,6 @@ type counterPacketBatchWriter struct { } func (w *counterPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } dataLens := make([]int64, len(buffers)) for index, buffer := range buffers { dataLens[index] = int64(buffer.Len()) @@ -120,10 +115,6 @@ type counterConnectedPacketBatchWriter struct { } func (w *counterConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers []*buf.Buffer) error { - if len(buffers) == 0 { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } dataLens := make([]int64, len(buffers)) for index, buffer := range buffers { dataLens[index] = int64(buffer.Len()) diff --git a/common/bufio/nat.go b/common/bufio/nat.go index e825ef46..12ffcbcb 100644 --- a/common/bufio/nat.go +++ b/common/bufio/nat.go @@ -3,7 +3,6 @@ package bufio import ( "net" "net/netip" - "os" "github.com/sagernet/sing/common/buf" M "github.com/sagernet/sing/common/metadata" @@ -86,10 +85,6 @@ type unidirectionalNATPacketBatchWriter struct { } func (w *unidirectionalNATPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } for index, destination := range destinations { if socksaddrWithoutPort(destination) == w.destination { destinations[index] = M.Socksaddr{ @@ -193,10 +188,6 @@ type bidirectionalNATPacketBatchWriter struct { } func (w *bidirectionalNATPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } for index, destination := range destinations { if socksaddrWithoutPort(destination) == w.destination { destinations[index] = M.Socksaddr{ @@ -281,10 +272,6 @@ type destinationNATPacketBatchWriter struct { } func (w *destinationNATPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } for index, destination := range destinations { if destination == w.destination { destinations[index] = w.origin diff --git a/common/bufio/nat_wait.go b/common/bufio/nat_wait.go index d6a6e2bd..94a7e05a 100644 --- a/common/bufio/nat_wait.go +++ b/common/bufio/nat_wait.go @@ -6,7 +6,15 @@ import ( N "github.com/sagernet/sing/common/network" ) -func (c *bidirectionalNATPacketConn) CreatePacketReadWaiter() (N.PacketReadWaiter, bool) { +func (c *unidirectionalNATPacketConn) CreateReadWaiter() (N.PacketReadWaiter, bool) { + return CreatePacketReadWaiter(c.NetPacketConn) +} + +func (c *unidirectionalNATPacketConn) CreatePacketBatchReadWaiter() (N.PacketBatchReadWaiter, bool) { + return CreatePacketBatchReadWaiter(c.NetPacketConn) +} + +func (c *bidirectionalNATPacketConn) CreateReadWaiter() (N.PacketReadWaiter, bool) { waiter, created := CreatePacketReadWaiter(c.NetPacketConn) if !created { return nil, false @@ -22,6 +30,22 @@ func (c *bidirectionalNATPacketConn) CreatePacketBatchReadWaiter() (N.PacketBatc return &batchWaitBidirectionalNATPacketConn{c, waiter}, true } +func (c *destinationNATPacketConn) CreateReadWaiter() (N.PacketReadWaiter, bool) { + waiter, created := CreatePacketReadWaiter(c.NetPacketConn) + if !created { + return nil, false + } + return &waitDestinationNATPacketConn{c, waiter}, true +} + +func (c *destinationNATPacketConn) CreatePacketBatchReadWaiter() (N.PacketBatchReadWaiter, bool) { + waiter, created := CreatePacketBatchReadWaiter(c.NetPacketConn) + if !created { + return nil, false + } + return &batchWaitDestinationNATPacketConn{c, waiter}, true +} + func (c *unidirectionalNATPacketConn) CreateConnectedPacketBatchReadWaiter() (N.ConnectedPacketBatchReadWaiter, bool) { return CreateConnectedPacketBatchReadWaiter(c.NetPacketConn) } @@ -66,6 +90,48 @@ func (c *waitBidirectionalNATPacketConn) WaitReadPacket() (buffer *buf.Buffer, d return } +type waitDestinationNATPacketConn struct { + *destinationNATPacketConn + readWaiter N.PacketReadWaiter +} + +func (c *waitDestinationNATPacketConn) InitializeReadWaiter(options N.ReadWaitOptions) (needCopy bool) { + return c.readWaiter.InitializeReadWaiter(options) +} + +func (c *waitDestinationNATPacketConn) WaitReadPacket() (buffer *buf.Buffer, destination M.Socksaddr, err error) { + buffer, destination, err = c.readWaiter.WaitReadPacket() + if err != nil { + return + } + if destination == c.origin { + destination = c.destination + } + return +} + +type batchWaitDestinationNATPacketConn struct { + *destinationNATPacketConn + readWaiter N.PacketBatchReadWaiter +} + +func (c *batchWaitDestinationNATPacketConn) InitializeReadWaiter(options N.ReadWaitOptions) (needCopy bool) { + return c.readWaiter.InitializeReadWaiter(options) +} + +func (c *batchWaitDestinationNATPacketConn) WaitReadPackets() (buffers []*buf.Buffer, destinations []M.Socksaddr, err error) { + buffers, destinations, err = c.readWaiter.WaitReadPackets() + if err != nil { + return + } + for index, destination := range destinations { + if destination == c.origin { + destinations[index] = c.destination + } + } + return +} + type connectedBatchWaitBidirectionalNATPacketConn struct { *bidirectionalNATPacketConn readWaiter N.ConnectedPacketBatchReadWaiter diff --git a/common/bufio/packet_batch.go b/common/bufio/packet_batch.go index a716e33d..a796c860 100644 --- a/common/bufio/packet_batch.go +++ b/common/bufio/packet_batch.go @@ -1,8 +1,6 @@ package bufio import ( - "os" - "github.com/sagernet/sing/common" "github.com/sagernet/sing/common/buf" M "github.com/sagernet/sing/common/metadata" @@ -71,10 +69,6 @@ type fallbackPacketBatchWriter struct { } func (w *fallbackPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if len(buffers) == 0 || len(buffers) != len(destinations) { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } for index, buffer := range buffers { err := w.writer.WritePacket(buffer, destinations[index]) if err != nil { @@ -91,10 +85,6 @@ type fallbackConnectedPacketBatchWriter struct { } func (w *fallbackConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers []*buf.Buffer) error { - if len(buffers) == 0 { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } for index, buffer := range buffers { err := w.writer.WritePacket(buffer, M.Socksaddr{}) if err != nil { diff --git a/common/bufio/packet_batch_mmsg.go b/common/bufio/packet_batch_mmsg.go index 6b820af5..710b710d 100644 --- a/common/bufio/packet_batch_mmsg.go +++ b/common/bufio/packet_batch_mmsg.go @@ -71,6 +71,7 @@ func createSyscallConnectedPacketBatchReadWaiter(reader any, destination M.Socks } func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) (needCopy bool) { + w.releaseBuffers() if options.BatchSize <= 0 { options.BatchSize = DefaultPacketReadBatchSize } @@ -108,9 +109,11 @@ func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOp case syscall.EINTR: continue case syscall.EAGAIN: + w.releaseBuffers() return false default: if errno == syscall.EWOULDBLOCK { + w.releaseBuffers() return false } w.readErr = os.NewSyscallError("recvmmsg", errno) @@ -134,12 +137,7 @@ func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOp } func (w *syscallPacketBatchReadWaiter) WaitReadPackets() (buffers []*buf.Buffer, destinations []M.Socksaddr, err error) { - if w.connected { - return nil, nil, os.ErrInvalid - } - if w.readFunc == nil { - return nil, nil, os.ErrInvalid - } + defer w.releaseBuffers() err = w.rawConn.Read(w.readFunc) if err != nil { return @@ -162,12 +160,7 @@ func (w *syscallPacketBatchReadWaiter) WaitReadPackets() (buffers []*buf.Buffer, } func (w *syscallPacketBatchReadWaiter) WaitReadConnectedPackets() (buffers []*buf.Buffer, destination M.Socksaddr, err error) { - if !w.connected { - return nil, M.Socksaddr{}, os.ErrInvalid - } - if w.readFunc == nil { - return nil, M.Socksaddr{}, os.ErrInvalid - } + defer w.releaseBuffers() err = w.rawConn.Read(w.readFunc) if err != nil { return @@ -194,9 +187,9 @@ var ( ) type syscallPacketBatchWriter struct { + offload syscallPacketBatchOffload upstream any rawConn syscall.RawConn - connected bool access sync.Mutex localAddr netip.AddrPort names []unix.RawSockaddrAny @@ -223,20 +216,13 @@ func createSyscallConnectedPacketBatchWriter(writer any) (N.ConnectedPacketBatch if _, isConnected := syscallPacketBatchPeerDestination(rawConn); !isConnected { return nil, false } - return &syscallPacketBatchWriter{upstream: writer, rawConn: rawConn, connected: true}, true + return &syscallPacketBatchWriter{upstream: writer, rawConn: rawConn}, true } func (w *syscallPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { - if w.connected { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } w.access.Lock() defer w.access.Unlock() defer buf.ReleaseMulti(buffers) - if len(buffers) == 0 || len(buffers) != len(destinations) { - return os.ErrInvalid - } if !w.localAddr.IsValid() { err := control.Raw(w.rawConn, func(fd uintptr) error { name, err := unix.Getsockname(int(fd)) @@ -254,6 +240,7 @@ func (w *syscallPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, desti iovecs := growSlice(w.iovecs, len(buffers)) msgvec := growSlice(w.msgvec, len(buffers)) defer func() { + w.offload.reset() clear(iovecs) clear(msgvec) w.names = names[:0] @@ -276,7 +263,7 @@ func (w *syscallPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, desti var innerErr syscall.Errno err := w.rawConn.Write(func(fd uintptr) (done bool) { for len(writeMsgvec) > 0 { - n, errno := sendmmsg(int(fd), writeMsgvec, 0) + n, errno := w.offload.send(int(fd), writeMsgvec, 0) switch errno { case 0: case syscall.EINTR: @@ -305,19 +292,13 @@ func (w *syscallPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, desti } func (w *syscallPacketBatchWriter) WriteConnectedPacketBatch(buffers []*buf.Buffer) error { - if !w.connected { - buf.ReleaseMulti(buffers) - return os.ErrInvalid - } w.access.Lock() defer w.access.Unlock() defer buf.ReleaseMulti(buffers) - if len(buffers) == 0 { - return os.ErrInvalid - } iovecs := growSlice(w.iovecs, len(buffers)) msgvec := growSlice(w.msgvec, len(buffers)) defer func() { + w.offload.reset() clear(iovecs) clear(msgvec) w.iovecs = iovecs[:0] @@ -336,7 +317,7 @@ func (w *syscallPacketBatchWriter) WriteConnectedPacketBatch(buffers []*buf.Buff var innerErr syscall.Errno err := w.rawConn.Write(func(fd uintptr) (done bool) { for len(writeMsgvec) > 0 { - n, errno := sendmmsg(int(fd), writeMsgvec, 0) + n, errno := w.offload.send(int(fd), writeMsgvec, 0) switch errno { case 0: case syscall.EINTR: diff --git a/common/bufio/packet_batch_msgx_darwin.go b/common/bufio/packet_batch_msgx_darwin.go index 0fb23e82..e771fce9 100644 --- a/common/bufio/packet_batch_msgx_darwin.go +++ b/common/bufio/packet_batch_msgx_darwin.go @@ -75,6 +75,7 @@ func createSyscallConnectedPacketBatchReadWaiter(reader any, destination M.Socks } func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) (needCopy bool) { + w.releaseBuffers() if options.BatchSize <= 0 { options.BatchSize = DefaultPacketReadBatchSize } @@ -113,9 +114,11 @@ func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOp case syscall.EINTR: continue case syscall.EAGAIN: + w.releaseBuffers() return false default: if errno == syscall.EWOULDBLOCK { + w.releaseBuffers() return false } w.readErr = os.NewSyscallError("recvmsg_x", errno) @@ -139,12 +142,7 @@ func (w *syscallPacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOp } func (w *syscallPacketBatchReadWaiter) WaitReadPackets() (buffers []*buf.Buffer, destinations []M.Socksaddr, err error) { - if w.connected { - return nil, nil, os.ErrInvalid - } - if w.readFunc == nil { - return nil, nil, os.ErrInvalid - } + defer w.releaseBuffers() err = w.rawConn.Read(w.readFunc) if err != nil { return @@ -167,12 +165,7 @@ func (w *syscallPacketBatchReadWaiter) WaitReadPackets() (buffers []*buf.Buffer, } func (w *syscallPacketBatchReadWaiter) WaitReadConnectedPackets() (buffers []*buf.Buffer, destination M.Socksaddr, err error) { - if !w.connected { - return nil, M.Socksaddr{}, os.ErrInvalid - } - if w.readFunc == nil { - return nil, M.Socksaddr{}, os.ErrInvalid - } + defer w.releaseBuffers() err = w.rawConn.Read(w.readFunc) if err != nil { return @@ -203,10 +196,6 @@ type syscallConnectedPacketBatchWriter struct { msgvec []msghdrX } -func createSyscallPacketBatchWriter(writer any) (N.PacketBatchWriter, bool) { - return nil, false -} - func createSyscallConnectedPacketBatchWriter(writer any) (N.ConnectedPacketBatchWriter, bool) { rawConn := syscallPacketBatchRawConnForWrite(writer) if rawConn == nil { @@ -222,9 +211,6 @@ func (w *syscallConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers [] w.access.Lock() defer w.access.Unlock() defer buf.ReleaseMulti(buffers) - if len(buffers) == 0 { - return os.ErrInvalid - } iovecs := growSlice(w.iovecs, len(buffers)) msgvec := growSlice(w.msgvec, len(buffers)) defer func() { @@ -245,10 +231,28 @@ func (w *syscallConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers [] writeMsgvec := msgvec maxBatchSize := len(writeMsgvec) var innerErr syscall.Errno + var innerErrName string err := w.rawConn.Write(func(fd uintptr) (done bool) { for len(writeMsgvec) > 0 { batchSize := min(maxBatchSize, len(writeMsgvec)) - n, errno := sendmsgX(int(fd), writeMsgvec[:batchSize], 0) + // The connected sendmsg_x path cannot send empty datagrams. Send + // those with sendto, batching the nonempty runs without reordering. + for index, message := range writeMsgvec[:batchSize] { + if message.iovlen == 0 { + batchSize = index + break + } + } + var n int + var errno syscall.Errno + syscallName := "sendmsg_x" + if batchSize == 0 { + syscallName = "sendto" + errno = sendto(int(fd), nil, nil, 0) + n = 1 + } else { + n, errno = sendmsgX(int(fd), writeMsgvec[:batchSize], 0) + } switch { case errno == 0: case errno == syscall.EINTR: @@ -260,10 +264,12 @@ func (w *syscallConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers [] return false default: innerErr = errno + innerErrName = syscallName return true } if n == 0 { innerErr = syscall.EIO + innerErrName = syscallName return true } writeMsgvec = writeMsgvec[n:] @@ -271,7 +277,7 @@ func (w *syscallConnectedPacketBatchWriter) WriteConnectedPacketBatch(buffers [] return true }) if innerErr != 0 { - err = os.NewSyscallError("sendmsg_x", innerErr) + err = os.NewSyscallError(innerErrName, innerErr) } return err } diff --git a/common/bufio/packet_batch_offload_linux.go b/common/bufio/packet_batch_offload_linux.go new file mode 100644 index 00000000..2c6c6ca4 --- /dev/null +++ b/common/bufio/packet_batch_offload_linux.go @@ -0,0 +1,87 @@ +package bufio + +import ( + "encoding/binary" + "syscall" + "unsafe" + + "golang.org/x/sys/unix" +) + +type syscallPacketBatchOffload struct { + disabled bool + messages []mmsghdr + controls [][24]byte + counts []int +} + +func (o *syscallPacketBatchOffload) reset() { + clear(o.messages[:cap(o.messages)]) +} + +func (o *syscallPacketBatchOffload) send(descriptor int, messages []mmsghdr, flags int) (int, syscall.Errno) { + if o.disabled || len(messages) < 2 { + return sendmmsg(descriptor, messages, flags) + } + o.messages = growSlice(o.messages, len(messages)) + o.controls = growSlice(o.controls, len(messages)) + o.counts = growSlice(o.counts, len(messages)) + messageCount := 0 + for start := 0; start < len(messages); { + first := messages[start].msgHdr + end := start + 1 + if first.Iovlen == 1 && first.Iov.Len > 0 { + segmentSize := int(first.Iov.Len) + length := segmentSize + for end < len(messages) && end-start < 64 { + next := messages[end].msgHdr + if next.Iovlen != 1 || next.Iov.Len == 0 || int(next.Iov.Len) > segmentSize || length+int(next.Iov.Len) > 65507 || next.Namelen != first.Namelen { + break + } + if first.Name != nil && *(*unix.RawSockaddrAny)(unsafe.Pointer(first.Name)) != *(*unix.RawSockaddrAny)(unsafe.Pointer(next.Name)) { + break + } + length += int(next.Iov.Len) + end++ + if int(next.Iov.Len) < segmentSize { + break + } + } + if end-start > 1 { + controlData := o.controls[messageCount][:unix.CmsgSpace(2)] + clear(controlData) + header := (*unix.Cmsghdr)(unsafe.Pointer(&controlData[0])) + header.Level = unix.IPPROTO_UDP + header.Type = unix.UDP_SEGMENT + header.SetLen(unix.CmsgLen(2)) + binary.NativeEndian.PutUint16(controlData[unix.CmsgLen(0):], uint16(segmentSize)) + first.Control = &controlData[0] + first.SetControllen(len(controlData)) + first.SetIovlen(end - start) + } + } + o.messages[messageCount] = mmsghdr{msgHdr: first} + o.counts[messageCount] = end - start + messageCount++ + start = end + } + if messageCount == len(messages) { + return sendmmsg(descriptor, messages, flags) + } + count, errno := sendmmsg(descriptor, o.messages[:messageCount], flags) + if errno != 0 { + switch errno { + case unix.EINVAL, unix.EIO, unix.ENOPROTOOPT, unix.EOPNOTSUPP: + o.disabled = true + return sendmmsg(descriptor, messages, flags) + case unix.EMSGSIZE: + return sendmmsg(descriptor, messages, flags) + } + return 0, errno + } + sent := 0 + for index := range count { + sent += o.counts[index] + } + return sent, 0 +} diff --git a/common/bufio/packet_batch_offload_netbsd.go b/common/bufio/packet_batch_offload_netbsd.go new file mode 100644 index 00000000..427d7000 --- /dev/null +++ b/common/bufio/packet_batch_offload_netbsd.go @@ -0,0 +1,11 @@ +package bufio + +import "syscall" + +type syscallPacketBatchOffload struct{} + +func (o *syscallPacketBatchOffload) reset() {} + +func (o *syscallPacketBatchOffload) send(descriptor int, messages []mmsghdr, flags int) (int, syscall.Errno) { + return sendmmsg(descriptor, messages, flags) +} diff --git a/common/bufio/packet_batch_read.go b/common/bufio/packet_batch_read.go new file mode 100644 index 00000000..f38dea0e --- /dev/null +++ b/common/bufio/packet_batch_read.go @@ -0,0 +1,13 @@ +//go:build linux || netbsd || darwin + +package bufio + +import "github.com/sagernet/sing/common/buf" + +func (w *syscallPacketBatchReadWaiter) releaseBuffers() { + clear(w.iovecs) + clear(w.msgvec) + buf.ReleaseMulti(w.buffers) + clear(w.buffers) + w.readN = 0 +} diff --git a/common/bufio/packet_batch_sendto_darwin.go b/common/bufio/packet_batch_sendto_darwin.go new file mode 100644 index 00000000..6e650067 --- /dev/null +++ b/common/bufio/packet_batch_sendto_darwin.go @@ -0,0 +1,107 @@ +package bufio + +import ( + "net/netip" + "os" + "sync" + "syscall" + "unsafe" + + "github.com/sagernet/sing/common/buf" + "github.com/sagernet/sing/common/control" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + + "golang.org/x/sys/unix" +) + +var _ N.PacketBatchWriter = (*syscallPacketBatchWriter)(nil) + +type syscallPacketBatchWriter struct { + upstream any + rawConn syscall.RawConn + access sync.Mutex + localAddr netip.AddrPort + names []unix.RawSockaddrAny + nameLens []uint32 +} + +func createSyscallPacketBatchWriter(writer any) (N.PacketBatchWriter, bool) { + rawConn := syscallPacketBatchRawConnForWrite(writer) + if rawConn == nil { + return nil, false + } + if _, isConnected := syscallPacketBatchPeerDestination(rawConn); isConnected { + return nil, false + } + return &syscallPacketBatchWriter{upstream: writer, rawConn: rawConn}, true +} + +func (w *syscallPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { + return w.writePacketBatch(buffers, destinations, sendto) +} + +func (w *syscallPacketBatchWriter) writePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr, send func(int, []byte, *unix.RawSockaddrAny, uint32) syscall.Errno) error { + w.access.Lock() + defer w.access.Unlock() + defer buf.ReleaseMulti(buffers) + if !w.localAddr.IsValid() { + err := control.Raw(w.rawConn, func(fd uintptr) error { + name, err := unix.Getsockname(int(fd)) + if err != nil { + return err + } + w.localAddr = M.AddrPortFromSockaddr(name) + return nil + }) + if err != nil { + return err + } + } + names := growSlice(w.names, len(buffers)) + nameLens := growSlice(w.nameLens, len(buffers)) + defer func() { + w.names = names[:0] + w.nameLens = nameLens[:0] + }() + for index, destination := range destinations { + nameLens[index] = M.AddrPortToRawSockaddrAny(&names[index], destination.AddrPort(), w.localAddr.Addr().Is6()) + } + // Keep the cursor across poller wakeups so a partially sent batch is not replayed. + var index int + var innerErr syscall.Errno + err := w.rawConn.Write(func(fd uintptr) (done bool) { + for index < len(buffers) { + errno := send(int(fd), buffers[index].Bytes(), &names[index], nameLens[index]) + switch errno { + case 0: + index++ + case syscall.EINTR: + continue + case syscall.EAGAIN: + return false + default: + if errno == syscall.EWOULDBLOCK { + return false + } + innerErr = errno + return true + } + } + return true + }) + if innerErr != 0 { + err = os.NewSyscallError("sendto", innerErr) + } + return err +} + +func (w *syscallPacketBatchWriter) Upstream() any { + return w.upstream +} + +func sendto(fd int, data []byte, name *unix.RawSockaddrAny, nameLen uint32) syscall.Errno { + //nolint:staticcheck + _, _, errno := unix.Syscall6(unix.SYS_SENDTO, uintptr(fd), uintptr(unsafe.Pointer(unsafe.SliceData(data))), uintptr(len(data)), 0, uintptr(unsafe.Pointer(name)), uintptr(nameLen)) + return errno +} diff --git a/common/bufio/packet_batch_sendto_darwin_test.go b/common/bufio/packet_batch_sendto_darwin_test.go new file mode 100644 index 00000000..8431d759 --- /dev/null +++ b/common/bufio/packet_batch_sendto_darwin_test.go @@ -0,0 +1,104 @@ +package bufio + +import ( + "net/netip" + "os" + "syscall" + "testing" + + M "github.com/sagernet/sing/common/metadata" + + "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" +) + +func TestPacketBatchSendtoResume(t *testing.T) { + t.Parallel() + for _, testCase := range []struct { + name string + errors []syscall.Errno + attempts []string + sent []string + pollErr error + wantErr error + wantWakeups int + }{ + { + name: "interrupted_and_blocked", + errors: []syscall.Errno{0, syscall.EINTR, syscall.EAGAIN, 0, 0}, + attempts: []string{"a", "b", "b", "b", "c"}, + sent: []string{"a", "b", "c"}, + wantWakeups: 2, + }, + { + name: "partial_error", + errors: []syscall.Errno{0, syscall.EMSGSIZE}, + attempts: []string{"a", "b"}, + sent: []string{"a"}, + wantErr: syscall.EMSGSIZE, + wantWakeups: 1, + }, + { + name: "partial_deadline", + errors: []syscall.Errno{0, syscall.EAGAIN}, + attempts: []string{"a", "b"}, + sent: []string{"a"}, + pollErr: os.ErrDeadlineExceeded, + wantErr: os.ErrDeadlineExceeded, + wantWakeups: 1, + }, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + rawConn := &testSendtoRawConn{pollErr: testCase.pollErr} + writer := &syscallPacketBatchWriter{rawConn: rawConn, localAddr: netip.MustParseAddrPort("127.0.0.1:1000")} + buffers := testBuffers("a", "b", "c") + destination := M.ParseSocksaddr("127.0.0.1:2000") + var attempts, sent []string + err := writer.writePacketBatch(buffers, []M.Socksaddr{destination, destination, destination}, func(_ int, data []byte, name *unix.RawSockaddrAny, nameLen uint32) syscall.Errno { + require.Less(t, len(attempts), len(testCase.errors), "unexpected send or retry") + require.EqualValues(t, unix.SizeofSockaddrInet4, nameLen) + require.Equal(t, destination, M.SocksaddrFromRawSockaddrAny(name)) + errno := testCase.errors[len(attempts)] + attempts = append(attempts, string(data)) + if errno == 0 { + sent = append(sent, string(data)) + } + return errno + }) + require.ErrorIs(t, err, testCase.wantErr) + require.Equal(t, testCase.attempts, attempts) + require.Equal(t, testCase.sent, sent) + require.Equal(t, testCase.wantWakeups, rawConn.wakeups) + for _, buffer := range buffers { + require.Zero(t, buffer.Cap()) + } + }) + } +} + +type testSendtoRawConn struct { + pollErr error + wakeups int +} + +func (c *testSendtoRawConn) Control(func(uintptr)) error { + panic("unexpected control") +} + +func (c *testSendtoRawConn) Read(func(uintptr) bool) error { + panic("unexpected read") +} + +func (c *testSendtoRawConn) Write(write func(uintptr) bool) error { + for range 4 { + c.wakeups++ + if write(0) { + return nil + } + if c.pollErr != nil { + return c.pollErr + } + } + return syscall.EIO +} diff --git a/common/bufio/packet_batch_syscall.go b/common/bufio/packet_batch_syscall.go index 4eaf18c7..825a71a7 100644 --- a/common/bufio/packet_batch_syscall.go +++ b/common/bufio/packet_batch_syscall.go @@ -14,37 +14,32 @@ import ( ) func syscallPacketBatchRawConnForRead(reader any) syscall.RawConn { + var rawConn syscall.RawConn if syscallConn, isSyscallConn := reader.(syscall.Conn); isSyscallConn { - rawConn, err := syscallConn.SyscallConn() - if err == nil { - return rawConn - } + rawConn, _ = syscallConn.SyscallConn() } - if ioReader, isReader := reader.(io.Reader); isReader { - _, rawConn := N.SyscallConnForRead(ioReader) - return rawConn + if rawConn == nil { + if ioReader, isReader := reader.(io.Reader); isReader { + _, rawConn = N.SyscallConnForRead(ioReader) + } } - return nil + return rawConn } func syscallPacketBatchRawConnForWrite(writer any) syscall.RawConn { + var rawConn syscall.RawConn if syscallConn, isSyscallConn := writer.(syscall.Conn); isSyscallConn { - rawConn, err := syscallConn.SyscallConn() - if err == nil { - return rawConn - } + rawConn, _ = syscallConn.SyscallConn() } - if ioWriter, isWriter := writer.(io.Writer); isWriter { - _, rawConn := N.SyscallConnForWrite(ioWriter) - return rawConn + if rawConn == nil { + if ioWriter, isWriter := writer.(io.Writer); isWriter { + _, rawConn = N.SyscallConnForWrite(ioWriter) + } } - return nil + return rawConn } func syscallPacketBatchPeerDestination(rawConn syscall.RawConn) (M.Socksaddr, bool) { - if rawConn == nil { - return M.Socksaddr{}, false - } var destination M.Socksaddr err := control.Raw(rawConn, func(fd uintptr) error { peer, err := unix.Getpeername(int(fd)) diff --git a/common/bufio/packet_batch_test.go b/common/bufio/packet_batch_test.go index a645186d..3237a7ef 100644 --- a/common/bufio/packet_batch_test.go +++ b/common/bufio/packet_batch_test.go @@ -5,6 +5,8 @@ import ( "io" "net" "net/netip" + "os" + "runtime" "strconv" "sync/atomic" "testing" @@ -297,6 +299,110 @@ func TestConnectedPacketBatchUDP(t *testing.T) { } } +func TestPacketBatchUDPMultipleDestinations(t *testing.T) { + t.Parallel() + for _, network := range []string{"udp4", "udp6"} { + t.Run(network, func(t *testing.T) { + t.Parallel() + input := listenPacketBatchUDP(t, network) + outputA := listenPacketBatchUDP(t, network) + outputB := listenPacketBatchUDP(t, network) + writer, created := CreatePacketBatchWriter(NewPacketConn(input)) + requirePacketBatchWriteBackend(t, created) + addressA := M.SocksaddrFromNet(outputA.LocalAddr()).Unwrap() + addressB := M.SocksaddrFromNet(outputB.LocalAddr()).Unwrap() + for range 2 { + require.NoError(t, writer.WritePacketBatch(testBuffers("a", "", "bc"), []M.Socksaddr{addressA, addressB, addressA})) + for _, payload := range []string{"a", "bc"} { + packet := make([]byte, 10) + n, _, err := outputA.ReadFromUDP(packet) + require.NoError(t, err) + require.Equal(t, payload, string(packet[:n])) + } + n, _, err := outputB.ReadFromUDP(make([]byte, 10)) + require.NoError(t, err) + require.Zero(t, n) + } + }) + } +} + +func TestPacketBatchUDPWriteErrors(t *testing.T) { + t.Parallel() + for _, connected := range []bool{false, true} { + t.Run(strconv.FormatBool(connected), func(t *testing.T) { + t.Parallel() + server := listenPacketBatchUDP(t, "udp4") + var conn *net.UDPConn + var write func([]*buf.Buffer) error + if connected { + var err error + conn, err = net.DialUDP("udp4", nil, server.LocalAddr().(*net.UDPAddr)) + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + writer, created := CreateConnectedPacketBatchWriter(NewUnbindPacketConn(conn)) + requirePacketBatchWriteBackend(t, created) + write = writer.WriteConnectedPacketBatch + } else { + conn = listenPacketBatchUDP(t, "udp4") + writer, created := CreatePacketBatchWriter(NewPacketConn(conn)) + requirePacketBatchWriteBackend(t, created) + write = func(buffers []*buf.Buffer) error { + destinations := make([]M.Socksaddr, len(buffers)) + for index := range destinations { + destinations[index] = M.SocksaddrFromNet(server.LocalAddr()).Unwrap() + } + return writer.WritePacketBatch(buffers, destinations) + } + } + require.NoError(t, write(testBuffers("", "a", "", "bc", ""))) + for _, payload := range []string{"", "a", "", "bc", ""} { + packet := make([]byte, 10) + n, _, err := server.ReadFromUDP(packet) + require.NoError(t, err) + require.Equal(t, payload, string(packet[:n])) + } + require.NoError(t, conn.SetWriteDeadline(time.Now().Add(-time.Second))) + buffers := testBuffers("deadline", "unsent") + require.ErrorIs(t, write(buffers), os.ErrDeadlineExceeded) + for _, buffer := range buffers { + require.Zero(t, buffer.Cap()) + } + require.NoError(t, conn.SetWriteDeadline(time.Time{})) + require.NoError(t, conn.Close()) + buffers = testBuffers("closed", "unsent") + require.ErrorIs(t, write(buffers), net.ErrClosed) + for _, buffer := range buffers { + require.Zero(t, buffer.Cap()) + } + }) + } +} + +func listenPacketBatchUDP(t *testing.T, network string) *net.UDPConn { + t.Helper() + ip := net.ParseIP("127.0.0.1") + if network == "udp6" { + ip = net.ParseIP("::1") + } + conn, err := net.ListenUDP(network, &net.UDPAddr{IP: ip}) + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + require.NoError(t, conn.SetDeadline(time.Now().Add(3*time.Second))) + return conn +} + +func requirePacketBatchWriteBackend(t *testing.T, created bool) { + t.Helper() + switch runtime.GOOS { + case "linux", "darwin", "netbsd", "android", "ios": + require.True(t, created) + default: + require.False(t, created) + t.Skip("packet batch backend is not available") + } +} + type testPacketBatch struct { payloads [][]byte destinations []M.Socksaddr diff --git a/common/control/socket_buffer.go b/common/control/socket_buffer.go new file mode 100644 index 00000000..9469bd75 --- /dev/null +++ b/common/control/socket_buffer.go @@ -0,0 +1,19 @@ +package control + +import ( + "syscall" + + N "github.com/sagernet/sing/common/network" +) + +func UDPSocketBuffer(size int) Func { + return func(network, address string, conn syscall.RawConn) error { + if N.NetworkName(network) != N.NetworkUDP { + return nil + } + return Raw(conn, func(fd uintptr) error { + setSocketBuffer(fd, size) + return nil + }) + } +} diff --git a/common/control/socket_buffer_linux.go b/common/control/socket_buffer_linux.go new file mode 100644 index 00000000..05e29e31 --- /dev/null +++ b/common/control/socket_buffer_linux.go @@ -0,0 +1,16 @@ +package control + +import ( + "golang.org/x/sys/unix" +) + +func setSocketBuffer(fd uintptr, size int) { + err := unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, size) + if err != nil { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, size) + } + err = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, size) + if err != nil { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUF, size) + } +} diff --git a/common/control/socket_buffer_other.go b/common/control/socket_buffer_other.go new file mode 100644 index 00000000..e3fb52d5 --- /dev/null +++ b/common/control/socket_buffer_other.go @@ -0,0 +1,6 @@ +//go:build !(unix || windows) + +package control + +func setSocketBuffer(fd uintptr, size int) { +} diff --git a/common/control/socket_buffer_unix.go b/common/control/socket_buffer_unix.go new file mode 100644 index 00000000..532e1202 --- /dev/null +++ b/common/control/socket_buffer_unix.go @@ -0,0 +1,12 @@ +//go:build unix && !linux + +package control + +import ( + "golang.org/x/sys/unix" +) + +func setSocketBuffer(fd uintptr, size int) { + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, size) + _ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUF, size) +} diff --git a/common/control/socket_buffer_windows.go b/common/control/socket_buffer_windows.go new file mode 100644 index 00000000..2d38b17a --- /dev/null +++ b/common/control/socket_buffer_windows.go @@ -0,0 +1,10 @@ +package control + +import ( + "golang.org/x/sys/windows" +) + +func setSocketBuffer(fd uintptr, size int) { + _ = windows.SetsockoptInt(windows.Handle(fd), windows.SOL_SOCKET, windows.SO_RCVBUF, size) + _ = windows.SetsockoptInt(windows.Handle(fd), windows.SOL_SOCKET, windows.SO_SNDBUF, size) +} diff --git a/common/domain/adguard_matcher.go b/common/domain/adguard_matcher.go index 5db165cf..e6946bb4 100644 --- a/common/domain/adguard_matcher.go +++ b/common/domain/adguard_matcher.go @@ -60,6 +60,18 @@ func ReadAdGuardMatcher(reader varbin.Reader) (*AdGuardMatcher, error) { return &AdGuardMatcher{set}, nil } +func NewAdGuardMatcherFromMmap(data Mmap) (*AdGuardMatcher, error) { + set, err := newSuccinctSetFromMmap(data) + if err != nil { + return nil, err + } + return &AdGuardMatcher{set}, nil +} + +func (m *AdGuardMatcher) Mmap() Mmap { + return m.set.mmap() +} + func (m *AdGuardMatcher) Write(writer varbin.Writer) error { return m.set.Write(writer) } diff --git a/common/domain/matcher.go b/common/domain/matcher.go index 679d1f14..d74c938a 100644 --- a/common/domain/matcher.go +++ b/common/domain/matcher.go @@ -56,6 +56,18 @@ func ReadMatcher(reader varbin.Reader) (*Matcher, error) { return &Matcher{set}, nil } +func NewMatcherFromMmap(data Mmap) (*Matcher, error) { + set, err := newSuccinctSetFromMmap(data) + if err != nil { + return nil, err + } + return &Matcher{set}, nil +} + +func (m *Matcher) Mmap() Mmap { + return m.set.mmap() +} + func (m *Matcher) Write(writer varbin.Writer) error { return m.set.Write(writer) } diff --git a/common/domain/set.go b/common/domain/set.go index 5683c2aa..c23b3731 100644 --- a/common/domain/set.go +++ b/common/domain/set.go @@ -14,6 +14,83 @@ type succinctSet struct { leaves, labelBitmap []uint64 labels []byte ranks, selects []int32 + storage any +} + +type Mmap struct { + Leaves []uint64 + LabelBitmap []uint64 + Labels []byte + Ranks []int32 + Selects []int32 + Storage any +} + +func (ss *succinctSet) mmap() Mmap { + return Mmap{ + Leaves: ss.leaves, + LabelBitmap: ss.labelBitmap, + Labels: ss.labels, + Ranks: ss.ranks, + Selects: ss.selects, + Storage: ss.storage, + } +} + +func newSuccinctSetFromMmap(data Mmap) (*succinctSet, error) { + onesCount, zerosCount := countLabelBitmap(data.LabelBitmap) + if onesCount != zerosCount+1 || len(data.Labels) != zerosCount { + return nil, E.New("domain: malformed succinct set") + } + if len(data.Leaves) < (onesCount+63)>>6 { + return nil, E.New("domain: malformed succinct set leaves") + } + if len(data.Ranks) != len(data.LabelBitmap)+1 { + return nil, E.New("domain: malformed succinct set ranks") + } + rank := int32(0) + for i, word := range data.LabelBitmap { + if data.Ranks[i] != rank { + return nil, E.New("domain: malformed succinct set ranks") + } + rank += int32(bits.OnesCount64(word)) + } + if data.Ranks[len(data.LabelBitmap)] != rank { + return nil, E.New("domain: malformed succinct set ranks") + } + if len(data.Selects) != (onesCount+31)>>5 { + return nil, E.New("domain: malformed succinct set selects") + } + ith := -1 + for i := range len(data.LabelBitmap) << 6 { + if data.LabelBitmap[i>>6]&(1<>5] != int32(i) { + return nil, E.New("domain: malformed succinct set selects") + } + } + return &succinctSet{ + leaves: data.Leaves, + labelBitmap: data.LabelBitmap, + labels: data.Labels, + ranks: data.Ranks, + selects: data.Selects, + storage: data.Storage, + }, nil +} + +func countLabelBitmap(labelBitmap []uint64) (onesCount int, zerosCount int) { + lastOneIndex := -1 + for wordIndex, word := range labelBitmap { + onesCount += bits.OnesCount64(word) + if word != 0 { + lastOneIndex = wordIndex<<6 | (63 - bits.LeadingZeros64(word)) + } + } + zerosCount = lastOneIndex + 1 - onesCount + return } func newSuccinctSet(keys []string) *succinctSet { @@ -92,15 +169,7 @@ func readSuccinctSet(reader varbin.Reader) (*succinctSet, error) { if err != nil { return nil, err } - onesCount := 0 - lastOneIndex := -1 - for wordIndex, word := range labelBitmap { - onesCount += bits.OnesCount64(word) - if word != 0 { - lastOneIndex = wordIndex<<6 | (63 - bits.LeadingZeros64(word)) - } - } - zerosCount := lastOneIndex + 1 - onesCount + onesCount, zerosCount := countLabelBitmap(labelBitmap) if onesCount != zerosCount+1 || len(labels) != zerosCount { return nil, E.New("domain: malformed succinct set") } diff --git a/common/memory/memory.go b/common/memory/memory.go index 54ca2492..8e4ec12d 100644 --- a/common/memory/memory.go +++ b/common/memory/memory.go @@ -18,6 +18,14 @@ func AvailableAvailable() bool { return availableAvailable() } +func Limit() uint64 { + return limitNative() +} + +func LimitAvailable() bool { + return limitAvailable() +} + func Inuse() uint64 { var memStats runtime.MemStats runtime.ReadMemStats(&memStats) diff --git a/common/memory/memory_darwin.go b/common/memory/memory_darwin.go index 86b0be2a..096482f1 100644 --- a/common/memory/memory_darwin.go +++ b/common/memory/memory_darwin.go @@ -30,7 +30,12 @@ package memory // return fn != NULL; // } import "C" -import "unsafe" + +import ( + "unsafe" + + "golang.org/x/sys/unix" +) func totalNative() uint64 { var vmInfo C.task_vm_info_data_t @@ -45,6 +50,18 @@ func totalAvailable() bool { return true } +func limitNative() uint64 { + size, err := unix.SysctlUint64("hw.memsize") + if err != nil { + return 0 + } + return size +} + +func limitAvailable() bool { + return true +} + func availableNative() uint64 { var supported C.int result := C.get_available_memory(&supported) diff --git a/common/memory/memory_linux.go b/common/memory/memory_linux.go index 64e65861..00954cf3 100644 --- a/common/memory/memory_linux.go +++ b/common/memory/memory_linux.go @@ -50,7 +50,7 @@ func availableNative() uint64 { if ok { return available } - return procMemAvailable() + return procMeminfo("MemAvailable:") } func availableAvailable() bool { @@ -66,24 +66,41 @@ func availableAvailable() bool { return true } -func cgroupAvailable() (uint64, bool) { - max, err := readCgroupUint("/sys/fs/cgroup/memory.max") - if err == nil && max != math.MaxUint64 { - current, err := readCgroupUint("/sys/fs/cgroup/memory.current") - if err == nil && max > current { - return max - current, true - } - return 0, true +func limitNative() uint64 { + total := procMeminfo("MemTotal:") + limit, _, found := cgroupLimit() + if found && (total == 0 || limit < total) { + return limit } - limit, err := readCgroupUint("/sys/fs/cgroup/memory/memory.limit_in_bytes") + return total +} + +func limitAvailable() bool { + return limitNative() != 0 +} + +func cgroupLimit() (uint64, string, bool) { + limit, err := readCgroupUint("/sys/fs/cgroup/memory.max") if err == nil && limit != math.MaxUint64 { - usage, err := readCgroupUint("/sys/fs/cgroup/memory/memory.usage_in_bytes") - if err == nil && limit > usage { - return limit - usage, true - } - return 0, true + return limit, "/sys/fs/cgroup/memory.current", true + } + limit, err = readCgroupUint("/sys/fs/cgroup/memory/memory.limit_in_bytes") + if err == nil && limit != math.MaxUint64 { + return limit, "/sys/fs/cgroup/memory/memory.usage_in_bytes", true + } + return 0, "", false +} + +func cgroupAvailable() (uint64, bool) { + limit, usagePath, found := cgroupLimit() + if !found { + return 0, false + } + usage, err := readCgroupUint(usagePath) + if err == nil && limit > usage { + return limit - usage, true } - return 0, false + return 0, true } func readCgroupUint(path string) (uint64, error) { @@ -98,7 +115,7 @@ func readCgroupUint(path string) (uint64, error) { return strconv.ParseUint(text, 10, 64) } -func procMemAvailable() uint64 { +func procMeminfo(field string) uint64 { file, err := os.Open("/proc/meminfo") if err != nil { return 0 @@ -107,16 +124,18 @@ func procMemAvailable() uint64 { scanner := bufio.NewScanner(file) for scanner.Scan() { line := scanner.Text() - if strings.HasPrefix(line, "MemAvailable:") { - fields := strings.Fields(line) - if len(fields) >= 2 { - value, err := strconv.ParseUint(fields[1], 10, 64) - if err == nil { - return value * 1024 - } - } - break + if !strings.HasPrefix(line, field) { + continue + } + fields := strings.Fields(line) + if len(fields) < 2 { + return 0 + } + value, parseErr := strconv.ParseUint(fields[1], 10, 64) + if parseErr != nil { + return 0 } + return value * 1024 } return 0 } diff --git a/common/memory/memory_stub.go b/common/memory/memory_stub.go index 24a4b050..add32f6a 100644 --- a/common/memory/memory_stub.go +++ b/common/memory/memory_stub.go @@ -10,6 +10,14 @@ func totalAvailable() bool { return false } +func limitNative() uint64 { + return 0 +} + +func limitAvailable() bool { + return false +} + func availableNative() uint64 { return 0 } diff --git a/common/memory/memory_windows.go b/common/memory/memory_windows.go index c0f3b0f0..020c5816 100644 --- a/common/memory/memory_windows.go +++ b/common/memory/memory_windows.go @@ -20,6 +20,20 @@ func totalAvailable() bool { return true } +func limitNative() uint64 { + var mem memoryStatusEx + mem.dwLength = uint32(unsafe.Sizeof(mem)) + err := globalMemoryStatusEx(&mem) + if err != nil { + return 0 + } + return mem.ullTotalPhys +} + +func limitAvailable() bool { + return true +} + func availableNative() uint64 { var mem memoryStatusEx mem.dwLength = uint32(unsafe.Sizeof(mem)) diff --git a/common/network/packet_offload.go b/common/network/packet_offload.go new file mode 100644 index 00000000..8b8c963a --- /dev/null +++ b/common/network/packet_offload.go @@ -0,0 +1,59 @@ +package network + +import ( + "syscall" + + "github.com/sagernet/sing/common" + "github.com/sagernet/sing/common/buf" + M "github.com/sagernet/sing/common/metadata" +) + +type PacketOffload interface { + EncodePacket(buffer *buf.Buffer, destination M.Socksaddr) error + DecodePacket(buffer *buf.Buffer) (M.Socksaddr, error) +} + +type PacketOffloadCreator interface { + CreatePacketOffload() (PacketOffload, bool) +} + +func UnwrapPacketOffload(conn any) (any, PacketOffload) { + var offload PacketOffload + for { + readerWithUpstream, isReaderWithUpstream := conn.(ReaderWithUpstream) + writerWithUpstream, isWriterWithUpstream := conn.(WriterWithUpstream) + replaceable := isReaderWithUpstream && readerWithUpstream.ReaderReplaceable() && isWriterWithUpstream && writerWithUpstream.WriterReplaceable() + creator, isCreator := conn.(PacketOffloadCreator) + if replaceable { + if _, isSyscallConn := conn.(syscall.Conn); isSyscallConn { + return conn, offload + } + } else if !isCreator || offload != nil { + return conn, offload + } + var upstream any + if withUpstream, hasUpstream := conn.(common.WithUpstream); hasUpstream { + upstream = withUpstream.Upstream() + } else { + upstreamReader, hasUpstreamReader := conn.(WithUpstreamReader) + upstreamWriter, hasUpstreamWriter := conn.(WithUpstreamWriter) + if hasUpstreamReader && hasUpstreamWriter { + upstream = upstreamReader.UpstreamReader() + if upstream != upstreamWriter.UpstreamWriter() { + upstream = nil + } + } + } + if upstream == nil { + return conn, offload + } + if !replaceable { + created, loaded := creator.CreatePacketOffload() + if !loaded { + return conn, offload + } + offload = created + } + conn = upstream + } +} diff --git a/common/winiphlpapi/helper.go b/common/winiphlpapi/helper.go index 25680b4d..3ea2e33c 100644 --- a/common/winiphlpapi/helper.go +++ b/common/winiphlpapi/helper.go @@ -8,7 +8,6 @@ import ( "net" "net/netip" "os" - "runtime" "slices" "syscall" "time" @@ -68,59 +67,151 @@ func LoadExtendedTable() error { if err != nil { return err } - return nil + err = procGetOwnerModuleFromTcpEntry.Find() + if err != nil { + return err + } + err = procGetOwnerModuleFromTcp6Entry.Find() + if err != nil { + return err + } + err = procGetOwnerModuleFromUdpEntry.Find() + if err != nil { + return err + } + err = procGetOwnerModuleFromUdp6Entry.Find() + if err != nil { + return err + } + // GetOwnerModuleFrom*Entry resolves service tags through advapi32 without loading it and fails with ERROR_MOD_NOT_FOUND until the process has. + err = procI_QueryTagInformation.Find() + if err != nil { + return err + } + return procNsiGetParameter.Find() +} + +type SocketOwner struct { + Pid uint32 + ServiceName string +} + +func FindPid(network string, source netip.AddrPort, destination netip.AddrPort) (uint32, error) { + owner, err := FindSocketOwner(network, source, destination) + if err != nil { + return 0, err + } + return owner.Pid, nil +} + +func FindSocketOwner(network string, source netip.AddrPort, destination netip.AddrPort) (SocketOwner, error) { + if N.NetworkName(network) == N.NetworkTCP && destination.IsValid() { + owner, found := findTCPSocketOwnerKeyed(source, destination) + if found { + return owner, nil + } + } + return findSocketOwnerInTable(network, source) +} + +func findTCPSocketOwnerKeyed(source netip.AddrPort, destination netip.AddrPort) (SocketOwner, bool) { + connection, err := nsiGetTCPConnection(source, destination) + if err != nil { + return SocketOwner{}, false + } + if source.Addr().Is4() { + row := MibTcpRowOwnerModule{DwOwningPid: connection.OwningPid} + row.OwningModuleInfo[0] = connection.OwningModuleInfo + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromTcpEntry(&row) + }), true + } + row := MibTcp6RowOwnerModule{DwOwningPid: connection.OwningPid} + row.OwningModuleInfo[0] = connection.OwningModuleInfo + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromTcp6Entry(&row) + }), true } -func FindPid(network string, source netip.AddrPort) (uint32, error) { +func findSocketOwnerInTable(network string, source netip.AddrPort) (SocketOwner, error) { switch N.NetworkName(network) { case N.NetworkTCP: if source.Addr().Is4() { - tcpTable, err := GetExtendedTcpTable() + tcpTable, err := GetExtendedTcpTableOwnerModule() if err != nil { - return 0, err + return SocketOwner{}, err } - for _, row := range tcpTable { - if source == netip.AddrPortFrom(DwordToAddr(row.DwLocalAddr), DwordToPort(row.DwLocalPort)) { - return row.DwOwningPid, nil - } + index := slices.IndexFunc(tcpTable, func(row MibTcpRowOwnerModule) bool { + return source == netip.AddrPortFrom(DwordToAddr(row.DwLocalAddr), DwordToPort(row.DwLocalPort)) + }) + if index != -1 { + row := &tcpTable[index] + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromTcpEntry(row) + }), nil } } else { - tcpTable, err := GetExtendedTcp6Table() + tcpTable, err := GetExtendedTcp6TableOwnerModule() if err != nil { - return 0, err + return SocketOwner{}, err } - for _, row := range tcpTable { - if source == netip.AddrPortFrom(netip.AddrFrom16(row.UcLocalAddr), DwordToPort(row.DwLocalPort)) { - return row.DwOwningPid, nil - } + index := slices.IndexFunc(tcpTable, func(row MibTcp6RowOwnerModule) bool { + return source == netip.AddrPortFrom(netip.AddrFrom16(row.UcLocalAddr), DwordToPort(row.DwLocalPort)) + }) + if index != -1 { + row := &tcpTable[index] + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromTcp6Entry(row) + }), nil } } case N.NetworkUDP: if source.Addr().Is4() { - udpTable, err := GetExtendedUdpTable() + udpTable, err := GetExtendedUdpTableOwnerModule() if err != nil { - return 0, err + return SocketOwner{}, err } - for _, row := range udpTable { - if source == netip.AddrPortFrom(DwordToAddr(row.DwLocalAddr), DwordToPort(row.DwLocalPort)) || - runtime.GOOS == "windows" && DwordToAddr(row.DwLocalAddr) == netip.IPv4Unspecified() && source.Port() == DwordToPort(row.DwLocalPort) { - return row.DwOwningPid, nil - } + index := slices.IndexFunc(udpTable, func(row MibUdpRowOwnerModule) bool { + return source == netip.AddrPortFrom(DwordToAddr(row.DwLocalAddr), DwordToPort(row.DwLocalPort)) || + DwordToAddr(row.DwLocalAddr) == netip.IPv4Unspecified() && source.Port() == DwordToPort(row.DwLocalPort) + }) + if index != -1 { + row := &udpTable[index] + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromUdpEntry(row) + }), nil } } else { - udpTable, err := GetExtendedUdp6Table() + udpTable, err := GetExtendedUdp6TableOwnerModule() if err != nil { - return 0, err + return SocketOwner{}, err } - for _, row := range udpTable { - if source == netip.AddrPortFrom(netip.AddrFrom16(row.UcLocalAddr), DwordToPort(row.DwLocalPort)) || - runtime.GOOS == "windows" && netip.AddrFrom16(row.UcLocalAddr) == netip.IPv6Unspecified() && source.Port() == DwordToPort(row.DwLocalPort) { - return row.DwOwningPid, nil - } + index := slices.IndexFunc(udpTable, func(row MibUdp6RowOwnerModule) bool { + return source == netip.AddrPortFrom(netip.AddrFrom16(row.UcLocalAddr), DwordToPort(row.DwLocalPort)) || + netip.AddrFrom16(row.UcLocalAddr) == netip.IPv6Unspecified() && source.Port() == DwordToPort(row.DwLocalPort) + }) + if index != -1 { + row := &udpTable[index] + return socketOwner(row.DwOwningPid, row.OwningModuleInfo[0], func() (*TcpipOwnerModuleBasicInfo, error) { + return GetOwnerModuleFromUdp6Entry(row) + }), nil } } } - return 0, E.New("process not found for ", source) + return SocketOwner{}, E.New("process not found for ", source) +} + +func socketOwner(pid uint32, serviceTag uint64, queryModule func() (*TcpipOwnerModuleBasicInfo, error)) SocketOwner { + owner := SocketOwner{Pid: pid} + if serviceTag == 0 { + return owner + } + moduleInfo, err := queryModule() + if err != nil { + return owner + } + owner.ServiceName = moduleInfo.ModuleName + return owner } func WriteAndWaitAck(ctx context.Context, conn net.Conn, payload []byte) error { diff --git a/common/winiphlpapi/iphlpapi.go b/common/winiphlpapi/iphlpapi.go index 74e5b90e..f53bf9dc 100644 --- a/common/winiphlpapi/iphlpapi.go +++ b/common/winiphlpapi/iphlpapi.go @@ -5,6 +5,7 @@ package winiphlpapi import ( "errors" "os" + "runtime" "unsafe" "golang.org/x/sys/windows" @@ -122,6 +123,65 @@ type MibUdp6RowOwnerPid struct { DwOwningPid uint32 } +type MibTcpRowOwnerModule struct { + DwState uint32 + DwLocalAddr uint32 + DwLocalPort uint32 + DwRemoteAddr uint32 + DwRemotePort uint32 + DwOwningPid uint32 + LiCreateTimestamp int64 + OwningModuleInfo [16]uint64 +} + +type MibTcp6RowOwnerModule struct { + UcLocalAddr [16]byte + DwLocalScopeId uint32 + DwLocalPort uint32 + UcRemoteAddr [16]byte + DwRemoteScopeId uint32 + DwRemotePort uint32 + DwState uint32 + DwOwningPid uint32 + LiCreateTimestamp int64 + OwningModuleInfo [16]uint64 +} + +type MibUdpRowOwnerModule struct { + DwLocalAddr uint32 + DwLocalPort uint32 + DwOwningPid uint32 + _ uint32 + LiCreateTimestamp int64 + DwFlags int32 + _ uint32 + OwningModuleInfo [16]uint64 +} + +type MibUdp6RowOwnerModule struct { + UcLocalAddr [16]byte + DwLocalScopeId uint32 + DwLocalPort uint32 + DwOwningPid uint32 + _ uint32 + LiCreateTimestamp int64 + DwFlags int32 + _ uint32 + OwningModuleInfo [16]uint64 +} + +const TcpipOwnerModuleInfoBasic uint32 = 0 + +type TcpipOwnerModuleBasicInfo struct { + ModuleName string + ModulePath string +} + +type tcpipOwnerModuleBasicInfo struct { + pModuleName *uint16 + pModulePath *uint16 +} + type TcpEstatsSendBufferRodV0 struct { CurRetxQueue uint64 MaxRetxQueue uint64 @@ -140,6 +200,7 @@ const ( offsetOfMibTcp6TableOwnerPid = unsafe.Offsetof(MibTcpTableOwnerPid{}.Table) offsetOfMibUdpTableOwnerPid = unsafe.Offsetof(MibUdpTableOwnerPid{}.Table) offsetOfMibUdp6TableOwnerPid = unsafe.Offsetof(MibUdp6TableOwnerPid{}.Table) + offsetOfMibTableOwnerModule = 8 sizeOfTcpEstatsSendBuffRwV0 = unsafe.Sizeof(TcpEstatsSendBuffRwV0{}) sizeOfTcpEstatsSendBufferRodV0 = unsafe.Sizeof(TcpEstatsSendBufferRodV0{}) ) @@ -264,6 +325,130 @@ func GetExtendedUdp6Table() ([]MibUdp6RowOwnerPid, error) { } } +func GetExtendedTcpTableOwnerModule() ([]MibTcpRowOwnerModule, error) { + var size uint32 + err := getExtendedTcpTable(nil, &size, false, windows.AF_INET, TcpTableOwnerModuleConnections, 0) + if !errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + return nil, os.NewSyscallError("GetExtendedTcpTable", err) + } + for { + table := make([]byte, size) + err = getExtendedTcpTable(&table[0], &size, false, windows.AF_INET, TcpTableOwnerModuleConnections, 0) + if err != nil { + if errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + continue + } + return nil, os.NewSyscallError("GetExtendedTcpTable", err) + } + dwNumEntries := int(*(*uint32)(unsafe.Pointer(&table[0]))) + return unsafe.Slice((*MibTcpRowOwnerModule)(unsafe.Pointer(&table[offsetOfMibTableOwnerModule])), dwNumEntries), nil + } +} + +func GetExtendedTcp6TableOwnerModule() ([]MibTcp6RowOwnerModule, error) { + var size uint32 + err := getExtendedTcpTable(nil, &size, false, windows.AF_INET6, TcpTableOwnerModuleConnections, 0) + if !errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + return nil, os.NewSyscallError("GetExtendedTcpTable", err) + } + for { + table := make([]byte, size) + err = getExtendedTcpTable(&table[0], &size, false, windows.AF_INET6, TcpTableOwnerModuleConnections, 0) + if err != nil { + if errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + continue + } + return nil, os.NewSyscallError("GetExtendedTcpTable", err) + } + dwNumEntries := int(*(*uint32)(unsafe.Pointer(&table[0]))) + return unsafe.Slice((*MibTcp6RowOwnerModule)(unsafe.Pointer(&table[offsetOfMibTableOwnerModule])), dwNumEntries), nil + } +} + +func GetExtendedUdpTableOwnerModule() ([]MibUdpRowOwnerModule, error) { + var size uint32 + err := getExtendedUdpTable(nil, &size, false, windows.AF_INET, UdpTableOwnerModule, 0) + if !errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + return nil, os.NewSyscallError("GetExtendedUdpTable", err) + } + for { + table := make([]byte, size) + err = getExtendedUdpTable(&table[0], &size, false, windows.AF_INET, UdpTableOwnerModule, 0) + if err != nil { + if errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + continue + } + return nil, os.NewSyscallError("GetExtendedUdpTable", err) + } + dwNumEntries := int(*(*uint32)(unsafe.Pointer(&table[0]))) + return unsafe.Slice((*MibUdpRowOwnerModule)(unsafe.Pointer(&table[offsetOfMibTableOwnerModule])), dwNumEntries), nil + } +} + +func GetExtendedUdp6TableOwnerModule() ([]MibUdp6RowOwnerModule, error) { + var size uint32 + err := getExtendedUdpTable(nil, &size, false, windows.AF_INET6, UdpTableOwnerModule, 0) + if !errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + return nil, os.NewSyscallError("GetExtendedUdpTable", err) + } + for { + table := make([]byte, size) + err = getExtendedUdpTable(&table[0], &size, false, windows.AF_INET6, UdpTableOwnerModule, 0) + if err != nil { + if errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + continue + } + return nil, os.NewSyscallError("GetExtendedUdpTable", err) + } + dwNumEntries := int(*(*uint32)(unsafe.Pointer(&table[0]))) + return unsafe.Slice((*MibUdp6RowOwnerModule)(unsafe.Pointer(&table[offsetOfMibTableOwnerModule])), dwNumEntries), nil + } +} + +func GetOwnerModuleFromTcpEntry(row *MibTcpRowOwnerModule) (*TcpipOwnerModuleBasicInfo, error) { + return queryOwnerModuleBasicInfo("GetOwnerModuleFromTcpEntry", func(buffer *byte, size *uint32) error { + return getOwnerModuleFromTcpEntry(row, TcpipOwnerModuleInfoBasic, buffer, size) + }) +} + +func GetOwnerModuleFromTcp6Entry(row *MibTcp6RowOwnerModule) (*TcpipOwnerModuleBasicInfo, error) { + return queryOwnerModuleBasicInfo("GetOwnerModuleFromTcp6Entry", func(buffer *byte, size *uint32) error { + return getOwnerModuleFromTcp6Entry(row, TcpipOwnerModuleInfoBasic, buffer, size) + }) +} + +func GetOwnerModuleFromUdpEntry(row *MibUdpRowOwnerModule) (*TcpipOwnerModuleBasicInfo, error) { + return queryOwnerModuleBasicInfo("GetOwnerModuleFromUdpEntry", func(buffer *byte, size *uint32) error { + return getOwnerModuleFromUdpEntry(row, TcpipOwnerModuleInfoBasic, buffer, size) + }) +} + +func GetOwnerModuleFromUdp6Entry(row *MibUdp6RowOwnerModule) (*TcpipOwnerModuleBasicInfo, error) { + return queryOwnerModuleBasicInfo("GetOwnerModuleFromUdp6Entry", func(buffer *byte, size *uint32) error { + return getOwnerModuleFromUdp6Entry(row, TcpipOwnerModuleInfoBasic, buffer, size) + }) +} + +func queryOwnerModuleBasicInfo(name string, query func(buffer *byte, size *uint32) error) (*TcpipOwnerModuleBasicInfo, error) { + var size uint32 + err := query(nil, &size) + if !errors.Is(err, windows.ERROR_INSUFFICIENT_BUFFER) { + return nil, os.NewSyscallError(name, err) + } + buffer := make([]byte, size) + err = query(&buffer[0], &size) + if err != nil { + return nil, os.NewSyscallError(name, err) + } + rawInfo := (*tcpipOwnerModuleBasicInfo)(unsafe.Pointer(&buffer[0])) + info := &TcpipOwnerModuleBasicInfo{ + ModuleName: windows.UTF16PtrToString(rawInfo.pModuleName), + ModulePath: windows.UTF16PtrToString(rawInfo.pModulePath), + } + runtime.KeepAlive(buffer) + return info, nil +} + func GetPerTcpConnectionEStatsSendBuffer(row *MibTcpRow) (*TcpEstatsSendBufferRodV0, error) { var rod TcpEstatsSendBufferRodV0 err := getPerTcpConnectionEStats(row, diff --git a/common/winiphlpapi/iphlpapi_test.go b/common/winiphlpapi/iphlpapi_test.go index 5e180977..e3b40565 100644 --- a/common/winiphlpapi/iphlpapi_test.go +++ b/common/winiphlpapi/iphlpapi_test.go @@ -5,6 +5,7 @@ package winiphlpapi_test import ( "context" "net" + "net/netip" "syscall" "testing" @@ -24,7 +25,7 @@ func TestFindPidTcp4(t *testing.T) { conn, err := net.Dial("tcp", listener.Addr().String()) require.NoError(t, err) defer conn.Close() - pid, err := winiphlpapi.FindPid(N.NetworkTCP, M.AddrPortFromNet(conn.LocalAddr())) + pid, err := winiphlpapi.FindPid(N.NetworkTCP, M.AddrPortFromNet(conn.LocalAddr()), netip.AddrPort{}) require.NoError(t, err) require.Equal(t, uint32(syscall.Getpid()), pid) } @@ -38,7 +39,7 @@ func TestFindPidTcp6(t *testing.T) { conn, err := net.Dial("tcp", listener.Addr().String()) require.NoError(t, err) defer conn.Close() - pid, err := winiphlpapi.FindPid(N.NetworkTCP, M.AddrPortFromNet(conn.LocalAddr())) + pid, err := winiphlpapi.FindPid(N.NetworkTCP, M.AddrPortFromNet(conn.LocalAddr()), netip.AddrPort{}) require.NoError(t, err) require.Equal(t, uint32(syscall.Getpid()), pid) } @@ -49,7 +50,7 @@ func TestFindPidUdp4(t *testing.T) { require.NoError(t, err) defer conn.Close() conn.Write([]byte("test")) - pid, err := winiphlpapi.FindPid(N.NetworkUDP, M.AddrPortFromNet(conn.LocalAddr())) + pid, err := winiphlpapi.FindPid(N.NetworkUDP, M.AddrPortFromNet(conn.LocalAddr()), netip.AddrPort{}) require.NoError(t, err) require.Equal(t, uint32(syscall.Getpid()), pid) } @@ -60,7 +61,7 @@ func TestFindPidUdp6(t *testing.T) { require.NoError(t, err) defer conn.Close() conn.Write([]byte("test")) - pid, err := winiphlpapi.FindPid(N.NetworkUDP, M.AddrPortFromNet(conn.LocalAddr())) + pid, err := winiphlpapi.FindPid(N.NetworkUDP, M.AddrPortFromNet(conn.LocalAddr()), netip.AddrPort{}) require.NoError(t, err) require.Equal(t, uint32(syscall.Getpid()), pid) } diff --git a/common/winiphlpapi/nsi.go b/common/winiphlpapi/nsi.go new file mode 100644 index 00000000..1e9c2ed7 --- /dev/null +++ b/common/winiphlpapi/nsi.go @@ -0,0 +1,76 @@ +//go:build windows + +package winiphlpapi + +import ( + "encoding/binary" + "net/netip" + "unsafe" + + "golang.org/x/sys/windows" +) + +// NsiGetParameter is undocumented; the TCP connection table index and the key and static +// parameter layouts below come from Wine (include/wine/nsi.h). The connection table is the +// one GetExtendedTcpTable reads for TCP_TABLE_*_CONNECTIONS, keyed by +// {SOCKADDR_INET local; SOCKADDR_INET remote}, and its static parameter carries the same +// owning pid, create timestamp and owning module info as MIB_TCPROW_OWNER_MODULE. +// The UDP endpoint table rejects parameter reads with ERROR_NOT_SUPPORTED. + +const ( + nsiStoreActive = 1 + nsiParameterStatic = 2 + nsiTCPConnectionTable = 4 + npiModuleIDTypeGUID = 1 + sockaddrInetSize = 28 +) + +type npiModuleID struct { + Length uint16 + _ uint16 + Type uint32 + GUID windows.GUID +} + +var npiTCPModuleID = npiModuleID{ + Length: uint16(unsafe.Sizeof(npiModuleID{})), + Type: npiModuleIDTypeGUID, + GUID: windows.GUID{ + Data1: 0xeb004a03, + Data2: 0x9b1a, + Data3: 0x11d4, + Data4: [8]byte{0x91, 0x23, 0x00, 0x50, 0x04, 0x77, 0x59, 0xbc}, + }, +} + +type nsiTCPConnectionStatic struct { + _ [3]uint32 + OwningPid uint32 + CreateTimestamp int64 + OwningModuleInfo uint64 +} + +func nsiGetTCPConnection(source netip.AddrPort, destination netip.AddrPort) (*nsiTCPConnectionStatic, error) { + var key [2 * sockaddrInetSize]byte + writeSockaddrInet(key[:sockaddrInetSize], source) + writeSockaddrInet(key[sockaddrInetSize:], destination) + var static nsiTCPConnectionStatic + err := nsiGetParameter(nsiStoreActive, &npiTCPModuleID, nsiTCPConnectionTable, &key[0], uint32(len(key)), nsiParameterStatic, unsafe.Pointer(&static), uint32(unsafe.Sizeof(static)), 0) + if err != nil { + return nil, err + } + return &static, nil +} + +func writeSockaddrInet(buffer []byte, addrPort netip.AddrPort) { + binary.BigEndian.PutUint16(buffer[2:], addrPort.Port()) + if addrPort.Addr().Is4() { + binary.NativeEndian.PutUint16(buffer, windows.AF_INET) + address := addrPort.Addr().As4() + copy(buffer[4:], address[:]) + } else { + binary.NativeEndian.PutUint16(buffer, windows.AF_INET6) + address := addrPort.Addr().As16() + copy(buffer[8:], address[:]) + } +} diff --git a/common/winiphlpapi/nsi_test.go b/common/winiphlpapi/nsi_test.go new file mode 100644 index 00000000..40181e48 --- /dev/null +++ b/common/winiphlpapi/nsi_test.go @@ -0,0 +1,72 @@ +//go:build windows + +package winiphlpapi + +import ( + "net" + "net/netip" + "testing" + + M "github.com/sagernet/sing/common/metadata" + + "github.com/stretchr/testify/require" +) + +func dialLoopback(t *testing.T, network string, address string) (netip.AddrPort, netip.AddrPort) { + listener, err := net.Listen(network, address) + require.NoError(t, err) + t.Cleanup(func() { listener.Close() }) + go func() { + for { + accepted, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + t.Cleanup(func() { accepted.Close() }) + } + }() + conn, err := net.Dial(network, listener.Addr().String()) + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + return M.AddrPortFromNet(conn.LocalAddr()), M.AddrPortFromNet(conn.RemoteAddr()) +} + +func TestNSITCPConnectionMatchesTable(t *testing.T) { + require.NoError(t, LoadExtendedTable()) + require.NoError(t, procNsiGetParameter.Find()) + dialLoopback(t, "tcp4", "127.0.0.1:0") + dialLoopback(t, "tcp6", "[::1]:0") + var checked int + table4, err := GetExtendedTcpTableOwnerModule() + require.NoError(t, err) + for _, row := range table4 { + local := netip.AddrPortFrom(DwordToAddr(row.DwLocalAddr), DwordToPort(row.DwLocalPort)) + remote := netip.AddrPortFrom(DwordToAddr(row.DwRemoteAddr), DwordToPort(row.DwRemotePort)) + connection, lookupErr := nsiGetTCPConnection(local, remote) + if lookupErr != nil { + continue + } + require.Equal(t, row.DwOwningPid, connection.OwningPid, "%v -> %v", local, remote) + require.Equal(t, row.LiCreateTimestamp, connection.CreateTimestamp, "%v -> %v", local, remote) + require.Equal(t, row.OwningModuleInfo[0], connection.OwningModuleInfo, "%v -> %v", local, remote) + checked++ + } + table6, err := GetExtendedTcp6TableOwnerModule() + require.NoError(t, err) + for _, row := range table6 { + if row.DwLocalScopeId != 0 || row.DwRemoteScopeId != 0 { + continue + } + local := netip.AddrPortFrom(netip.AddrFrom16(row.UcLocalAddr), DwordToPort(row.DwLocalPort)) + remote := netip.AddrPortFrom(netip.AddrFrom16(row.UcRemoteAddr), DwordToPort(row.DwRemotePort)) + connection, lookupErr := nsiGetTCPConnection(local, remote) + if lookupErr != nil { + continue + } + require.Equal(t, row.DwOwningPid, connection.OwningPid, "%v -> %v", local, remote) + require.Equal(t, row.LiCreateTimestamp, connection.CreateTimestamp, "%v -> %v", local, remote) + require.Equal(t, row.OwningModuleInfo[0], connection.OwningModuleInfo, "%v -> %v", local, remote) + checked++ + } + require.GreaterOrEqual(t, checked, 4) +} diff --git a/common/winiphlpapi/syscall_windows.go b/common/winiphlpapi/syscall_windows.go index f6aab14c..a89288b7 100644 --- a/common/winiphlpapi/syscall_windows.go +++ b/common/winiphlpapi/syscall_windows.go @@ -25,3 +25,20 @@ package winiphlpapi // https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getextendedudptable //sys getExtendedUdpTable(pUdpTable *byte, pdwSize *uint32, bOrder bool, ulAf uint64, tableClass uint32, reserved uint64) = (errcode error) = iphlpapi.GetExtendedUdpTable + +// https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getownermodulefromtcpentry +//sys getOwnerModuleFromTcpEntry(pTcpEntry *MibTcpRowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) = iphlpapi.GetOwnerModuleFromTcpEntry + +// https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getownermodulefromtcp6entry +//sys getOwnerModuleFromTcp6Entry(pTcpEntry *MibTcp6RowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) = iphlpapi.GetOwnerModuleFromTcp6Entry + +// https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getownermodulefromudpentry +//sys getOwnerModuleFromUdpEntry(pUdpEntry *MibUdpRowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) = iphlpapi.GetOwnerModuleFromUdpEntry + +// https://learn.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getownermodulefromudp6entry +//sys getOwnerModuleFromUdp6Entry(pUdpEntry *MibUdp6RowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) = iphlpapi.GetOwnerModuleFromUdp6Entry + +//sys queryTagInformation(machineName *uint16, infoLevel uint32, tagInfo unsafe.Pointer) (errcode error) = advapi32.I_QueryTagInformation + +// Undocumented; signature from Wine dlls/nsi/nsi.c +//sys nsiGetParameter(store uint32, module *npiModuleID, table uint32, key *byte, keySize uint32, parameterType uint32, data unsafe.Pointer, dataSize uint32, dataOffset uint32) (errcode error) = nsi.NsiGetParameter diff --git a/common/winiphlpapi/zsyscall_windows.go b/common/winiphlpapi/zsyscall_windows.go index e5e93088..69dbdf54 100644 --- a/common/winiphlpapi/zsyscall_windows.go +++ b/common/winiphlpapi/zsyscall_windows.go @@ -38,24 +38,40 @@ func errnoErr(e syscall.Errno) error { } var ( + modadvapi32 = windows.NewLazySystemDLL("advapi32.dll") modiphlpapi = windows.NewLazySystemDLL("iphlpapi.dll") - - procGetExtendedTcpTable = modiphlpapi.NewProc("GetExtendedTcpTable") - procGetExtendedUdpTable = modiphlpapi.NewProc("GetExtendedUdpTable") - procGetPerTcp6ConnectionEStats = modiphlpapi.NewProc("GetPerTcp6ConnectionEStats") - procGetPerTcpConnectionEStats = modiphlpapi.NewProc("GetPerTcpConnectionEStats") - procGetTcp6Table = modiphlpapi.NewProc("GetTcp6Table") - procGetTcpTable = modiphlpapi.NewProc("GetTcpTable") - procSetPerTcp6ConnectionEStats = modiphlpapi.NewProc("SetPerTcp6ConnectionEStats") - procSetPerTcpConnectionEStats = modiphlpapi.NewProc("SetPerTcpConnectionEStats") + modnsi = windows.NewLazySystemDLL("nsi.dll") + + procI_QueryTagInformation = modadvapi32.NewProc("I_QueryTagInformation") + procGetExtendedTcpTable = modiphlpapi.NewProc("GetExtendedTcpTable") + procGetExtendedUdpTable = modiphlpapi.NewProc("GetExtendedUdpTable") + procGetOwnerModuleFromTcp6Entry = modiphlpapi.NewProc("GetOwnerModuleFromTcp6Entry") + procGetOwnerModuleFromTcpEntry = modiphlpapi.NewProc("GetOwnerModuleFromTcpEntry") + procGetOwnerModuleFromUdp6Entry = modiphlpapi.NewProc("GetOwnerModuleFromUdp6Entry") + procGetOwnerModuleFromUdpEntry = modiphlpapi.NewProc("GetOwnerModuleFromUdpEntry") + procGetPerTcp6ConnectionEStats = modiphlpapi.NewProc("GetPerTcp6ConnectionEStats") + procGetPerTcpConnectionEStats = modiphlpapi.NewProc("GetPerTcpConnectionEStats") + procGetTcp6Table = modiphlpapi.NewProc("GetTcp6Table") + procGetTcpTable = modiphlpapi.NewProc("GetTcpTable") + procSetPerTcp6ConnectionEStats = modiphlpapi.NewProc("SetPerTcp6ConnectionEStats") + procSetPerTcpConnectionEStats = modiphlpapi.NewProc("SetPerTcpConnectionEStats") + procNsiGetParameter = modnsi.NewProc("NsiGetParameter") ) +func queryTagInformation(machineName *uint16, infoLevel uint32, tagInfo unsafe.Pointer) (errcode error) { + r0, _, _ := syscall.SyscallN(procI_QueryTagInformation.Addr(), uintptr(unsafe.Pointer(machineName)), uintptr(infoLevel), uintptr(tagInfo)) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + func getExtendedTcpTable(pTcpTable *byte, pdwSize *uint32, bOrder bool, ulAf uint64, tableClass uint32, reserved uint64) (errcode error) { var _p0 uint32 if bOrder { _p0 = 1 } - r0, _, _ := syscall.Syscall6(procGetExtendedTcpTable.Addr(), 6, uintptr(unsafe.Pointer(pTcpTable)), uintptr(unsafe.Pointer(pdwSize)), uintptr(_p0), uintptr(ulAf), uintptr(tableClass), uintptr(reserved)) + r0, _, _ := syscall.SyscallN(procGetExtendedTcpTable.Addr(), uintptr(unsafe.Pointer(pTcpTable)), uintptr(unsafe.Pointer(pdwSize)), uintptr(_p0), uintptr(ulAf), uintptr(tableClass), uintptr(reserved)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -67,7 +83,39 @@ func getExtendedUdpTable(pUdpTable *byte, pdwSize *uint32, bOrder bool, ulAf uin if bOrder { _p0 = 1 } - r0, _, _ := syscall.Syscall6(procGetExtendedUdpTable.Addr(), 6, uintptr(unsafe.Pointer(pUdpTable)), uintptr(unsafe.Pointer(pdwSize)), uintptr(_p0), uintptr(ulAf), uintptr(tableClass), uintptr(reserved)) + r0, _, _ := syscall.SyscallN(procGetExtendedUdpTable.Addr(), uintptr(unsafe.Pointer(pUdpTable)), uintptr(unsafe.Pointer(pdwSize)), uintptr(_p0), uintptr(ulAf), uintptr(tableClass), uintptr(reserved)) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + +func getOwnerModuleFromTcp6Entry(pTcpEntry *MibTcp6RowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) { + r0, _, _ := syscall.SyscallN(procGetOwnerModuleFromTcp6Entry.Addr(), uintptr(unsafe.Pointer(pTcpEntry)), uintptr(class), uintptr(unsafe.Pointer(pBuffer)), uintptr(unsafe.Pointer(pdwSize))) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + +func getOwnerModuleFromTcpEntry(pTcpEntry *MibTcpRowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) { + r0, _, _ := syscall.SyscallN(procGetOwnerModuleFromTcpEntry.Addr(), uintptr(unsafe.Pointer(pTcpEntry)), uintptr(class), uintptr(unsafe.Pointer(pBuffer)), uintptr(unsafe.Pointer(pdwSize))) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + +func getOwnerModuleFromUdp6Entry(pUdpEntry *MibUdp6RowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) { + r0, _, _ := syscall.SyscallN(procGetOwnerModuleFromUdp6Entry.Addr(), uintptr(unsafe.Pointer(pUdpEntry)), uintptr(class), uintptr(unsafe.Pointer(pBuffer)), uintptr(unsafe.Pointer(pdwSize))) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + +func getOwnerModuleFromUdpEntry(pUdpEntry *MibUdpRowOwnerModule, class uint32, pBuffer *byte, pdwSize *uint32) (errcode error) { + r0, _, _ := syscall.SyscallN(procGetOwnerModuleFromUdpEntry.Addr(), uintptr(unsafe.Pointer(pUdpEntry)), uintptr(class), uintptr(unsafe.Pointer(pBuffer)), uintptr(unsafe.Pointer(pdwSize))) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -75,7 +123,7 @@ func getExtendedUdpTable(pUdpTable *byte, pdwSize *uint32, bOrder bool, ulAf uin } func getPerTcp6ConnectionEStats(row *MibTcp6Row, estatsType uint32, rw uintptr, rwVersion uint64, rwSize uint64, ros uintptr, rosVersion uint64, rosSize uint64, rod uintptr, rodVersion uint64, rodSize uint64) (errcode error) { - r0, _, _ := syscall.Syscall12(procGetPerTcp6ConnectionEStats.Addr(), 11, uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(ros), uintptr(rosVersion), uintptr(rosSize), uintptr(rod), uintptr(rodVersion), uintptr(rodSize), 0) + r0, _, _ := syscall.SyscallN(procGetPerTcp6ConnectionEStats.Addr(), uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(ros), uintptr(rosVersion), uintptr(rosSize), uintptr(rod), uintptr(rodVersion), uintptr(rodSize)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -83,7 +131,7 @@ func getPerTcp6ConnectionEStats(row *MibTcp6Row, estatsType uint32, rw uintptr, } func getPerTcpConnectionEStats(row *MibTcpRow, estatsType uint32, rw uintptr, rwVersion uint64, rwSize uint64, ros uintptr, rosVersion uint64, rosSize uint64, rod uintptr, rodVersion uint64, rodSize uint64) (errcode error) { - r0, _, _ := syscall.Syscall12(procGetPerTcpConnectionEStats.Addr(), 11, uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(ros), uintptr(rosVersion), uintptr(rosSize), uintptr(rod), uintptr(rodVersion), uintptr(rodSize), 0) + r0, _, _ := syscall.SyscallN(procGetPerTcpConnectionEStats.Addr(), uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(ros), uintptr(rosVersion), uintptr(rosSize), uintptr(rod), uintptr(rodVersion), uintptr(rodSize)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -95,7 +143,7 @@ func getTcp6Table(tcpTable *byte, sizePointer *uint32, order bool) (errcode erro if order { _p0 = 1 } - r0, _, _ := syscall.Syscall(procGetTcp6Table.Addr(), 3, uintptr(unsafe.Pointer(tcpTable)), uintptr(unsafe.Pointer(sizePointer)), uintptr(_p0)) + r0, _, _ := syscall.SyscallN(procGetTcp6Table.Addr(), uintptr(unsafe.Pointer(tcpTable)), uintptr(unsafe.Pointer(sizePointer)), uintptr(_p0)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -107,7 +155,7 @@ func getTcpTable(tcpTable *byte, sizePointer *uint32, order bool) (errcode error if order { _p0 = 1 } - r0, _, _ := syscall.Syscall(procGetTcpTable.Addr(), 3, uintptr(unsafe.Pointer(tcpTable)), uintptr(unsafe.Pointer(sizePointer)), uintptr(_p0)) + r0, _, _ := syscall.SyscallN(procGetTcpTable.Addr(), uintptr(unsafe.Pointer(tcpTable)), uintptr(unsafe.Pointer(sizePointer)), uintptr(_p0)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -115,7 +163,7 @@ func getTcpTable(tcpTable *byte, sizePointer *uint32, order bool) (errcode error } func setPerTcp6ConnectionEStats(row *MibTcp6Row, estatsType uint32, rw uintptr, rwVersion uint64, rwSize uint64, offset uint64) (errcode error) { - r0, _, _ := syscall.Syscall6(procSetPerTcp6ConnectionEStats.Addr(), 6, uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(offset)) + r0, _, _ := syscall.SyscallN(procSetPerTcp6ConnectionEStats.Addr(), uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(offset)) if r0 != 0 { errcode = syscall.Errno(r0) } @@ -123,7 +171,15 @@ func setPerTcp6ConnectionEStats(row *MibTcp6Row, estatsType uint32, rw uintptr, } func setPerTcpConnectionEStats(row *MibTcpRow, estatsType uint32, rw uintptr, rwVersion uint64, rwSize uint64, offset uint64) (errcode error) { - r0, _, _ := syscall.Syscall6(procSetPerTcpConnectionEStats.Addr(), 6, uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(offset)) + r0, _, _ := syscall.SyscallN(procSetPerTcpConnectionEStats.Addr(), uintptr(unsafe.Pointer(row)), uintptr(estatsType), uintptr(rw), uintptr(rwVersion), uintptr(rwSize), uintptr(offset)) + if r0 != 0 { + errcode = syscall.Errno(r0) + } + return +} + +func nsiGetParameter(store uint32, module *npiModuleID, table uint32, key *byte, keySize uint32, parameterType uint32, data unsafe.Pointer, dataSize uint32, dataOffset uint32) (errcode error) { + r0, _, _ := syscall.SyscallN(procNsiGetParameter.Addr(), uintptr(store), uintptr(unsafe.Pointer(module)), uintptr(table), uintptr(unsafe.Pointer(key)), uintptr(keySize), uintptr(parameterType), uintptr(data), uintptr(dataSize), uintptr(dataOffset)) if r0 != 0 { errcode = syscall.Errno(r0) } diff --git a/protocol/socks/handshake.go b/protocol/socks/handshake.go index c6970ea3..77f4dcb4 100644 --- a/protocol/socks/handshake.go +++ b/protocol/socks/handshake.go @@ -258,10 +258,18 @@ func HandleConnectionEx( if udpTimeout > 0 { udpConn.SetReadDeadline(time.Now().Add(udpTimeout)) } - firstPacket := buf.NewPacket() + var firstPacket *buf.Buffer var destination M.Socksaddr - destination, err = socksPacketConn.ReadPacket(firstPacket) + readWaiter, hasReadWaiter := bufio.CreatePacketReadWaiter(socksPacketConn) + if hasReadWaiter { + readWaiter.InitializeReadWaiter(N.ReadWaitOptions{}) + firstPacket, destination, err = readWaiter.WaitReadPacket() + } else { + firstPacket = buf.NewPacket() + destination, err = socksPacketConn.ReadPacket(firstPacket) + } if err != nil { + firstPacket.Release() _ = socksPacketConn.Close() return E.Cause(err, "socks5: read first packet") } diff --git a/protocol/socks/lazy.go b/protocol/socks/lazy.go index 1a1d61ca..bfac68f8 100644 --- a/protocol/socks/lazy.go +++ b/protocol/socks/lazy.go @@ -5,7 +5,6 @@ import ( "os" "github.com/sagernet/sing/common/buf" - "github.com/sagernet/sing/common/bufio" M "github.com/sagernet/sing/common/metadata" "github.com/sagernet/sing/protocol/socks/socks4" "github.com/sagernet/sing/protocol/socks/socks5" @@ -107,11 +106,7 @@ type LazyAssociatePacketConn struct { func NewLazyAssociatePacketConn(conn net.Conn, underlying net.Conn) *LazyAssociatePacketConn { return &LazyAssociatePacketConn{ - AssociatePacketConn: AssociatePacketConn{ - AbstractConn: conn, - conn: bufio.NewExtendedConn(conn), - underlying: underlying, - }, + AssociatePacketConn: *NewAssociatePacketConn(conn, M.Socksaddr{}, underlying), } } diff --git a/protocol/socks/packet.go b/protocol/socks/packet.go index 583600ca..0560ac72 100644 --- a/protocol/socks/packet.go +++ b/protocol/socks/packet.go @@ -94,6 +94,7 @@ func (c *AssociatePacketConn) WritePacket(buffer *buf.Buffer, destination M.Sock common.Must(header.WriteZeroN(3)) err := M.SocksaddrSerializer.WriteAddrPort(header, destination) if err != nil { + buffer.Release() return err } return c.conn.WriteBuffer(buffer) diff --git a/protocol/socks/packet_batch.go b/protocol/socks/packet_batch.go new file mode 100644 index 00000000..1c8e5039 --- /dev/null +++ b/protocol/socks/packet_batch.go @@ -0,0 +1,148 @@ +package socks + +import ( + "github.com/sagernet/sing/common" + "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" +) + +func (c *AssociatePacketConn) CreatePacketBatchWriter() (N.PacketBatchWriter, bool) { + writer, created := bufio.CreateConnectedPacketBatchWriter(bufio.NewUnbindPacketConn(c.conn)) + if !created { + return nil, false + } + return &associatePacketBatchWriter{writer: writer}, true +} + +type associatePacketBatchWriter struct { + writer N.ConnectedPacketBatchWriter +} + +func (c *AssociatePacketConn) CreatePacketBatchReadWaiter() (N.PacketBatchReadWaiter, bool) { + reader, created := bufio.CreateConnectedPacketBatchReadWaiter(bufio.NewUnbindPacketConn(c.conn)) + if !created { + return nil, false + } + return &associatePacketBatchReadWaiter{conn: c, reader: reader}, true +} + +type associatePacketBatchReadWaiter struct { + conn *AssociatePacketConn + reader N.ConnectedPacketBatchReadWaiter +} + +func (r *associatePacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) bool { + return r.reader.InitializeReadWaiter(options) +} + +func (r *associatePacketBatchReadWaiter) WaitReadPackets() ([]*buf.Buffer, []M.Socksaddr, error) { + buffers, _, err := r.reader.WaitReadConnectedPackets() + if err != nil { + return nil, nil, err + } + destinations := make([]M.Socksaddr, len(buffers)) + for index, buffer := range buffers { + destinations[index], err = (associatePacketOffload{}).DecodePacket(buffer) + if err != nil { + buf.ReleaseMulti(buffers) + return nil, nil, err + } + } + r.conn.remoteAddr = destinations[len(destinations)-1] + return buffers, destinations, nil +} + +func (r *associatePacketBatchReadWaiter) Upstream() any { + return r.reader +} + +func (c *LazyAssociatePacketConn) CreatePacketBatchReadWaiter() (N.PacketBatchReadWaiter, bool) { + reader, created := c.AssociatePacketConn.CreatePacketBatchReadWaiter() + if !created { + return nil, false + } + return &lazyAssociatePacketBatchReadWaiter{conn: c, reader: reader}, true +} + +type lazyAssociatePacketBatchReadWaiter struct { + conn *LazyAssociatePacketConn + reader N.PacketBatchReadWaiter +} + +func (r *lazyAssociatePacketBatchReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) bool { + return r.reader.InitializeReadWaiter(options) +} + +func (r *lazyAssociatePacketBatchReadWaiter) WaitReadPackets() ([]*buf.Buffer, []M.Socksaddr, error) { + err := r.conn.HandshakeSuccess() + if err != nil { + return nil, nil, err + } + return r.reader.WaitReadPackets() +} + +func (r *lazyAssociatePacketBatchReadWaiter) Upstream() any { return r.reader } + +func (w *associatePacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { + for index, buffer := range buffers { + destination := destinations[index] + headerLen := 3 + M.SocksaddrSerializer.AddrPortLen(destination) + if buffer.Start() < headerLen { + newBuffer := buf.NewSize(headerLen + buffer.Len()) + newBuffer.Resize(headerLen, 0) + common.Must1(newBuffer.Write(buffer.Bytes())) + buffer.Release() + buffer = newBuffer + buffers[index] = buffer + } + header := buf.With(buffer.ExtendHeader(headerLen)) + common.Must(header.WriteZeroN(3)) + err := M.SocksaddrSerializer.WriteAddrPort(header, destination) + if err != nil { + buf.ReleaseMulti(buffers) + return err + } + } + return w.writer.WriteConnectedPacketBatch(buffers) +} + +func (w *associatePacketBatchWriter) Upstream() any { + return w.writer +} + +func (c *LazyAssociatePacketConn) CreatePacketBatchWriter() (N.PacketBatchWriter, bool) { + writer, created := c.AssociatePacketConn.CreatePacketBatchWriter() + if !created { + return nil, false + } + return &lazyAssociatePacketBatchWriter{c, writer}, true +} + +type lazyAssociatePacketBatchWriter struct { + conn *LazyAssociatePacketConn + writer N.PacketBatchWriter +} + +func (w *lazyAssociatePacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error { + err := w.conn.HandshakeSuccess() + if err != nil { + buf.ReleaseMulti(buffers) + return err + } + return w.writer.WritePacketBatch(buffers, destinations) +} + +func (w *lazyAssociatePacketBatchWriter) Upstream() any { return w.writer } + +var ( + _ N.PacketBatchReadWaitCreator = (*AssociatePacketConn)(nil) + _ N.PacketBatchWriteCreator = (*AssociatePacketConn)(nil) + _ N.PacketBatchReadWaitCreator = (*LazyAssociatePacketConn)(nil) + _ N.PacketBatchWriteCreator = (*LazyAssociatePacketConn)(nil) + _ N.PacketBatchReadWaiter = (*associatePacketBatchReadWaiter)(nil) + _ N.PacketBatchWriter = (*associatePacketBatchWriter)(nil) + _ N.PacketBatchReadWaiter = (*lazyAssociatePacketBatchReadWaiter)(nil) + _ N.PacketBatchWriter = (*lazyAssociatePacketBatchWriter)(nil) +) diff --git a/protocol/socks/packet_offload.go b/protocol/socks/packet_offload.go new file mode 100644 index 00000000..d0c0c661 --- /dev/null +++ b/protocol/socks/packet_offload.go @@ -0,0 +1,43 @@ +package socks + +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" +) + +type associatePacketOffload struct{} + +func (c *AssociatePacketConn) CreatePacketOffload() (N.PacketOffload, bool) { + return associatePacketOffload{}, true +} + +func (c *LazyAssociatePacketConn) CreatePacketOffload() (N.PacketOffload, bool) { + return nil, false +} + +func (o associatePacketOffload) EncodePacket(buffer *buf.Buffer, destination M.Socksaddr) error { + headerLen := 3 + M.SocksaddrSerializer.AddrPortLen(destination) + header := buf.With(buffer.ExtendHeader(headerLen)) + common.Must(header.WriteZeroN(3)) + return M.SocksaddrSerializer.WriteAddrPort(header, destination) +} + +func (o associatePacketOffload) DecodePacket(buffer *buf.Buffer) (M.Socksaddr, error) { + if buffer.Len() < 3 { + return M.Socksaddr{}, ErrInvalidPacket + } + buffer.Advance(3) + destination, err := M.SocksaddrSerializer.ReadAddrPort(buffer) + if err != nil { + return M.Socksaddr{}, E.Cause1(ErrInvalidPacket, err) + } + return destination, nil +} + +var ( + _ N.PacketOffloadCreator = (*AssociatePacketConn)(nil) + _ N.PacketOffloadCreator = (*LazyAssociatePacketConn)(nil) +) diff --git a/protocol/socks/packet_vectorised.go b/protocol/socks/packet_vectorised.go index 6a4e5912..9b8939bf 100644 --- a/protocol/socks/packet_vectorised.go +++ b/protocol/socks/packet_vectorised.go @@ -19,27 +19,19 @@ type VectorisedAssociatePacketConn struct { func NewVectorisedAssociateConn(conn net.Conn, writer N.VectorisedWriter, remoteAddr M.Socksaddr, underlying net.Conn) *VectorisedAssociatePacketConn { return &VectorisedAssociatePacketConn{ - AssociatePacketConn{ - AbstractConn: conn, - conn: bufio.NewExtendedConn(conn), - remoteAddr: remoteAddr, - underlying: underlying, - }, + *NewAssociatePacketConn(conn, remoteAddr, underlying), &bufio.UnbindVectorisedPacketWriter{VectorisedWriter: writer}, } } func (c *VectorisedAssociatePacketConn) WriteVectorisedPacket(buffers []*buf.Buffer, destination M.Socksaddr) error { header := buf.NewSize(3 + M.SocksaddrSerializer.AddrPortLen(destination)) - defer header.Release() common.Must(header.WriteZeroN(3)) err := M.SocksaddrSerializer.WriteAddrPort(header, destination) if err != nil { + header.Release() + buf.ReleaseMulti(buffers) return err } return c.VectorisedPacketWriter.WriteVectorisedPacket(append([]*buf.Buffer{header}, buffers...), destination) } - -func (c *VectorisedAssociatePacketConn) FrontHeadroom() int { - return 0 -}