Add tcp-brutal support

server.go

@@ -5,6 +5,7 @@ import (
 	"net"
 
 	"github.com/sagernet/sing/common/bufio"
+	"github.com/sagernet/sing/common/debug"
 	E "github.com/sagernet/sing/common/exceptions"
 	"github.com/sagernet/sing/common/logger"
 	M "github.com/sagernet/sing/common/metadata"
@@ -12,19 +13,49 @@ import (
 	"github.com/sagernet/sing/common/task"
 )
 
-type ServerHandler interface {
+type ServiceHandler interface {
 	N.TCPConnectionHandler
 	N.UDPConnectionHandler
-	E.Handler
 }
 
-func HandleConnection(ctx context.Context, handler ServerHandler, logger logger.ContextLogger, conn net.Conn, metadata M.Metadata) error {
+type Service struct {
+	newStreamContext func(context.Context, net.Conn) context.Context
+	logger           logger.ContextLogger
+	handler          ServiceHandler
+	padding          bool
+	brutal           BrutalOptions
+}
+
+type ServiceOptions struct {
+	NewStreamContext func(context.Context, net.Conn) context.Context
+	Logger           logger.ContextLogger
+	Handler          ServiceHandler
+	Padding          bool
+	Brutal           BrutalOptions
+}
+
+func NewService(options ServiceOptions) (*Service, error) {
+	if options.Brutal.Enabled && !BrutalAvailable && !debug.Enabled {
+		return nil, E.New("TCP Brutal is only supported on Linux")
+	}
+	return &Service{
+		newStreamContext: options.NewStreamContext,
+		logger:           options.Logger,
+		handler:          options.Handler,
+		padding:          options.Padding,
+		brutal:           options.Brutal,
+	}, nil
+}
+
+func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
 	request, err := ReadRequest(conn)
 	if err != nil {
 		return err
 	}
 	if request.Padding {
 		conn = newPaddingConn(conn)
+	} else if s.padding {
+		return E.New("non-padded connection rejected")
 	}
 	session, err := newServerSession(conn, request.Protocol)
 	if err != nil {
@@ -38,7 +69,13 @@ func HandleConnection(ctx context.Context, handler ServerHandler, logger logger.
 			if err != nil {
 				return err
 			}
-			go newConnection(ctx, handler, logger, stream, metadata)
+			streamCtx := s.newStreamContext(ctx, stream)
+			go func() {
+				hErr := s.newConnection(streamCtx, conn, stream, metadata)
+				if hErr != nil {
+					s.logger.ErrorContext(streamCtx, E.Cause(hErr, "handle connection"))
+				}
+			}()
 		}
 	})
 	group.Cleanup(func() {
@@ -47,34 +84,64 @@ func HandleConnection(ctx context.Context, handler ServerHandler, logger logger.
 	return group.Run(ctx)
 }
 
-func newConnection(ctx context.Context, handler ServerHandler, logger logger.ContextLogger, stream net.Conn, metadata M.Metadata) {
+func (s *Service) newConnection(ctx context.Context, sessionConn net.Conn, stream net.Conn, metadata M.Metadata) error {
 	stream = &wrapStream{stream}
 	request, err := ReadStreamRequest(stream)
 	if err != nil {
-		logger.ErrorContext(ctx, err)
-		return
+		return E.Cause(err, "read multiplex stream request")
 	}
 	metadata.Destination = request.Destination
 	if request.Network == N.NetworkTCP {
-		logger.InfoContext(ctx, "inbound multiplex connection to ", metadata.Destination)
-		hErr := handler.NewConnection(ctx, &serverConn{ExtendedConn: bufio.NewExtendedConn(stream)}, metadata)
-		stream.Close()
-		if hErr != nil {
-			handler.NewError(ctx, hErr)
+		conn := &serverConn{ExtendedConn: bufio.NewExtendedConn(stream)}
+		if request.Destination.Fqdn == BrutalExchangeDomain {
+			defer stream.Close()
+			var clientReceiveBPS uint64
+			clientReceiveBPS, err = ReadBrutalRequest(conn)
+			if err != nil {
+				return E.Cause(err, "read brutal request")
+			}
+			if !s.brutal.Enabled {
+				err = WriteBrutalResponse(conn, 0, false, "brutal is not enabled by the server")
+				if err != nil {
+					return E.Cause(err, "write brutal response")
+				}
+				return nil
+			}
+			sendBPS := s.brutal.SendBPS
+			if clientReceiveBPS < sendBPS {
+				sendBPS = clientReceiveBPS
+			}
+			err = SetBrutalOptions(sessionConn, sendBPS)
+			if err != nil {
+				// ignore error in test
+				if !debug.Enabled {
+					err = WriteBrutalResponse(conn, 0, false, E.Cause(err, "enable TCP Brutal").Error())
+					if err != nil {
+						return E.Cause(err, "write brutal response")
+					}
+					return nil
+				}
+			}
+			err = WriteBrutalResponse(conn, s.brutal.ReceiveBPS, true, "")
+			if err != nil {
+				return E.Cause(err, "write brutal response")
+			}
+			return nil
 		}
+		s.logger.InfoContext(ctx, "inbound multiplex connection to ", metadata.Destination)
+		s.handler.NewConnection(ctx, conn, metadata)
+		stream.Close()
 	} else {
 		var packetConn N.PacketConn
 		if !request.PacketAddr {
-			logger.InfoContext(ctx, "inbound multiplex packet connection to ", metadata.Destination)
+			s.logger.InfoContext(ctx, "inbound multiplex packet connection to ", metadata.Destination)
 			packetConn = &serverPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: request.Destination}
 		} else {
-			logger.InfoContext(ctx, "inbound multiplex packet connection")
+			s.logger.InfoContext(ctx, "inbound multiplex packet connection")
 			packetConn = &serverPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream)}
 		}
-		hErr := handler.NewPacketConnection(ctx, packetConn, metadata)
+		s.handler.NewPacketConnection(ctx, packetConn, metadata)
 		stream.Close()
-		if hErr != nil {
-			handler.NewError(ctx, hErr)
-		}
 	}
+	return nil
 }