给 Xray 的 gRPC 添加 TCP Brutal 支持

infra/conf/transport_internet.go

@@ -649,6 +649,8 @@ type SocketConfig struct {
  V6only bool `json:"v6only"`
  Interface string `json:"interface"`
  TcpMptcp bool `json:"tcpMptcp"`
+ BrutalRate uint64 `json:"brutalRate"`
+ BrutalGain uint32 `json:"brutalGain"`
 }
 
 // Build implements Buildable.
@@ -723,6 +725,8 @@ func (c *SocketConfig) Build() (*internet.SocketConfig, error) {
  V6Only: c.V6only,
  Interface: c.Interface,
  TcpMptcp: c.TcpMptcp,
+ BrutalRate: c.BrutalRate,
+ BrutalGain: c.BrutalGain,
  }, nil
 }
 

transport/internet/config.proto

@@ -119,4 +119,8 @@ message SocketConfig {
  bool tcp_no_delay = 18;
 
  bool tcp_mptcp = 19;
+
+ uint64 brutal_rate = 20;
+
+ uint32 brutal_gain = 21;
 }

transport/internet/grpc/hub.go

@@ -3,6 +3,7 @@ package grpc
 import (
  "context"
  "time"
+ _ "unsafe"
 
  goreality "github.com/xtls/reality"
  "github.com/xtls/xray-core/common"
@@ -17,24 +18,36 @@ import (
  "google.golang.org/grpc/keepalive"
 )
 
+//go:linkname GetConnection google.golang.org/grpc/internal/transport.GetConnection
+func GetConnection(ctx context.Context) net.Conn
+
 type Listener struct {
  encoding.UnimplementedGRPCServiceServer
  ctx context.Context
  handler internet.ConnHandler
  local net.Addr
  config *Config
+ stream *internet.MemoryStreamConfig
 
  s *grpc.Server
 }
 
 func (l Listener) Tun(server encoding.GRPCService_TunServer) error {
+ if l.stream != nil {
+ conn := GetConnection(server.Context())
+ l.stream.ApplyBrutalSettings(conn)
+ }
  tunCtx, cancel := context.WithCancel(l.ctx)
  l.handler(encoding.NewHunkConn(server, cancel))
  <-tunCtx.Done()
  return nil
 }
 
 func (l Listener) TunMulti(server encoding.GRPCService_TunMultiServer) error {
+ if l.stream != nil {
+ conn := GetConnection(server.Context())
+ l.stream.ApplyBrutalSettings(conn)
+ }
  tunCtx, cancel := context.WithCancel(l.ctx)
  l.handler(encoding.NewMultiHunkConn(server, cancel))
  <-tunCtx.Done()
@@ -55,6 +68,7 @@ func Listen(ctx context.Context, address net.Address, port net.Port, settings *i
  var listener *Listener
  if port == net.Port(0) { // unix
  listener = &Listener{
+ stream: settings,
  handler: handler,
  local: &net.UnixAddr{
  Name: address.Domain(),
@@ -64,6 +78,7 @@ func Listen(ctx context.Context, address net.Address, port net.Port, settings *i
  }
  } else { // tcp
  listener = &Listener{
+ stream: settings,
  handler: handler,
  local: &net.TCPAddr{
  IP: address.IP(),

transport/internet/memory_settings.go

@@ -1,5 +1,11 @@
 package internet
 
+import (
+ "syscall"
+ sing_common "github.com/sagernet/sing/common"
+ "github.com/xtls/xray-core/common/net"
+)
+
 // MemoryStreamConfig is a parsed form of StreamConfig. This is used to reduce number of Protobuf parsing.
 type MemoryStreamConfig struct {
  ProtocolName string
@@ -9,6 +15,28 @@
  SocketSettings *SocketConfig
 }
 
+func (m *MemoryStreamConfig) ApplyBrutalSettings(conn net.Conn) {
+ if m.SocketSettings != nil && m.SocketSettings.TcpCongestion == "brutal" {
+ sc, loaded := sing_common.Cast[syscall.Conn](conn)
+ if loaded {
+ if rawConn, e0 := sc.SyscallConn(); e0 != nil {
+ newError("sc.SyscallConn failed ", e0).AtError().WriteToLog()
+ } else {
+ if e1 := rawConn.Control(func(fd uintptr) {
+ if e2 := ApplyBrutalParams(int(fd), m.SocketSettings); e2 != nil {
+ newError("internet.ApplyBrutalParams failed ", e2).AtError().WriteToLog()
+ }
+ }); e1 != nil {
+ newError("rawConn.Control failed ", e1).AtError().WriteToLog()
+ }
+ }
+ } else {
+ newError("cast conn to syscall.Conn failed").AtWarning().WriteToLog()
+ }
+ newError("Brutal Rate: ", m.SocketSettings.BrutalRate, ", Gain: ", m.SocketSettings.BrutalGain).WriteToLog()
+ }
+}
+
 // ToMemoryStreamConfig converts a StreamConfig to MemoryStreamConfig. It returns a default non-nil MemoryStreamConfig for nil input.
 func ToMemoryStreamConfig(s *StreamConfig) (*MemoryStreamConfig, error) {
  ets, err := s.GetEffectiveTransportSettings()

transport/internet/sockopt_linux.go

@@ -3,10 +3,19 @@ package internet
 import (
  "net"
  "syscall"
+ "unsafe"
 
  "golang.org/x/sys/unix"
 )
 
+type TCPBrutalParams struct {
+ Rate uint64
+ Gain uint32
+}
+
+//go:linkname setsockopt syscall.setsockopt
+func setsockopt(s, level, name int, val unsafe.Pointer, vallen uintptr) (err error)
+
 func bindAddr(fd uintptr, ip []byte, port uint32) error {
  setReuseAddr(fd)
  setReusePort(fd)
@@ -200,6 +209,18 @@ func applyInboundSocketOptions(network string, fd uintptr, config *SocketConfig)
  return nil
 }
 
+func ApplyBrutalParams(fd int, config *SocketConfig) error {
+ var params TCPBrutalParams
+
+ params.Rate = config.BrutalRate
+ params.Gain = config.BrutalGain
+
+ if err := setsockopt(fd, unix.IPPROTO_TCP, 23301, unsafe.Pointer(&params), unsafe.Sizeof(params)); err != nil {
+ return err
+ }
+ return nil
+}
+
 func setReuseAddr(fd uintptr) error {
  if err := syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1); err != nil {
  return newError("failed to set SO_REUSEADDR").Base(err).AtWarning()

transport/internet/tcp/hub.go

@@ -24,12 +24,14 @@ type Listener struct {
  authConfig internet.ConnectionAuthenticator
  config *Config
  addConn internet.ConnHandler
+ stream *internet.MemoryStreamConfig
 }
 
 // ListenTCP creates a new Listener based on configurations.
 func ListenTCP(ctx context.Context, address net.Address, port net.Port, streamSettings *internet.MemoryStreamConfig, handler internet.ConnHandler) (internet.Listener, error) {
  l := &Listener{
  addConn: handler,
+ stream: streamSettings,
  }
  tcpSettings := streamSettings.ProtocolSettings.(*Config)
  l.config = tcpSettings
@@ -104,6 +106,11 @@ func (v *Listener) keepAccepting() {
  }
  continue
  }
+
+ if v.stream != nil {
+ v.stream.ApplyBrutalSettings(conn)
+ }
+
  go func() {
  if v.tlsConfig != nil {
  conn = tls.Server(conn, v.tlsConfig)