mirror of
https://github.com/SagerNet/sing-mux.git
synced 2025-04-01 19:17:36 +03:00
Update network handler usages
This commit is contained in:
世界
parent
5c2a0c4fa6
commit
54badfa885
6 changed files with 68 additions and 29 deletions
|
|
@ -7,7 +7,7 @@ import (
|
|||
"github.com/sagernet/sing/common"
|
||||
"github.com/sagernet/sing/common/buf"
|
||||
E "github.com/sagernet/sing/common/exceptions"
|
||||
"github.com/sagernet/sing/common/rw"
|
||||
"github.com/sagernet/sing/common/varbin"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -32,7 +32,7 @@ func WriteBrutalResponse(writer io.Writer, receiveBPS uint64, ok bool, message s
|
|||
if ok {
|
||||
common.Must(binary.Write(buffer, binary.BigEndian, receiveBPS))
|
||||
} else {
|
||||
err := rw.WriteVString(buffer, message)
|
||||
err := varbin.Write(buffer, binary.BigEndian, message)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -52,7 +52,7 @@ func ReadBrutalResponse(reader io.Reader) (uint64, error) {
|
|||
return receiveBPS, err
|
||||
} else {
|
||||
var message string
|
||||
message, err = rw.ReadVString(reader)
|
||||
message, err = varbin.ReadValue[string](reader, binary.BigEndian)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
|
|
|||
2
go.mod
2
go.mod
|
|
@ -4,7 +4,7 @@ go 1.18
|
|||
|
||||
require (
|
||||
github.com/hashicorp/yamux v0.1.2
|
||||
github.com/sagernet/sing v0.5.0
|
||||
github.com/sagernet/sing v0.6.0-alpha.3
|
||||
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7
|
||||
golang.org/x/net v0.31.0
|
||||
golang.org/x/sys v0.27.0
|
||||
|
|
|
|||
6
go.sum
6
go.sum
|
|
@ -3,8 +3,10 @@ github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8
|
|||
github.com/hashicorp/yamux v0.1.2/go.mod h1:C+zze2n6e/7wshOZep2A70/aQU6QBRWJO/G6FT1wIns=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/sagernet/sing v0.2.18/go.mod h1:OL6k2F0vHmEzXz2KW19qQzu172FDgSbUSODylighuVo=
|
||||
github.com/sagernet/sing v0.5.0 h1:soo2wVwLcieKWWKIksFNK6CCAojUgAppqQVwyRYGkEM=
|
||||
github.com/sagernet/sing v0.5.0/go.mod h1:ARkL0gM13/Iv5VCZmci/NuoOlePoIsW0m7BWfln/Hak=
|
||||
github.com/sagernet/sing v0.5.1-0.20241109034027-099899991126 h1:pLMpV9pEAinrS9R1n1JLcbNesCl369RfvyxnYCPrkbw=
|
||||
github.com/sagernet/sing v0.5.1-0.20241109034027-099899991126/go.mod h1:ARkL0gM13/Iv5VCZmci/NuoOlePoIsW0m7BWfln/Hak=
|
||||
github.com/sagernet/sing v0.6.0-alpha.3 h1:GLp9d6Gbt+Ioeplauuzojz1nY2J6moceVGYIOv/h5gA=
|
||||
github.com/sagernet/sing v0.6.0-alpha.3/go.mod h1:ARkL0gM13/Iv5VCZmci/NuoOlePoIsW0m7BWfln/Hak=
|
||||
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7 h1:DImB4lELfQhplLTxeq2z31Fpv8CQqqrUwTbrIRumZqQ=
|
||||
github.com/sagernet/smux v0.0.0-20231208180855-7041f6ea79e7/go.mod h1:FP9X2xjT/Az1EsG/orYYoC+5MojWnuI7hrffz8fGwwo=
|
||||
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
|
||||
|
|
|
|||
16
protocol.go
16
protocol.go
|
|
@ -12,6 +12,7 @@ import (
|
|||
M "github.com/sagernet/sing/common/metadata"
|
||||
N "github.com/sagernet/sing/common/network"
|
||||
"github.com/sagernet/sing/common/rw"
|
||||
"github.com/sagernet/sing/common/varbin"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -41,14 +42,18 @@ type Request struct {
|
|||
}
|
||||
|
||||
func ReadRequest(reader io.Reader) (*Request, error) {
|
||||
version, err := rw.ReadByte(reader)
|
||||
var (
|
||||
version byte
|
||||
protocol byte
|
||||
)
|
||||
err := binary.Read(reader, binary.BigEndian, &version)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if version < Version0 || version > Version1 {
|
||||
return nil, E.New("unsupported version: ", version)
|
||||
}
|
||||
protocol, err := rw.ReadByte(reader)
|
||||
err = binary.Read(reader, binary.BigEndian, &protocol)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -166,13 +171,12 @@ type StreamResponse struct {
|
|||
|
||||
func ReadStreamResponse(reader io.Reader) (*StreamResponse, error) {
|
||||
var response StreamResponse
|
||||
status, err := rw.ReadByte(reader)
|
||||
err := binary.Read(reader, binary.BigEndian, &response.Status)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
response.Status = status
|
||||
if status == statusError {
|
||||
response.Message, err = rw.ReadVString(reader)
|
||||
if response.Status == statusError {
|
||||
response.Message, err = varbin.ReadValue[string](reader, binary.BigEndian)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
53
server.go
53
server.go
|
|
@ -13,15 +13,24 @@ import (
|
|||
"github.com/sagernet/sing/common/task"
|
||||
)
|
||||
|
||||
// Deprecated: Use ServiceHandlerEx instead.
|
||||
//
|
||||
//nolint:staticcheck
|
||||
type ServiceHandler interface {
|
||||
N.TCPConnectionHandler
|
||||
N.UDPConnectionHandler
|
||||
}
|
||||
|
||||
type ServiceHandlerEx interface {
|
||||
N.TCPConnectionHandlerEx
|
||||
N.UDPConnectionHandlerEx
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
newStreamContext func(context.Context, net.Conn) context.Context
|
||||
logger logger.ContextLogger
|
||||
handler ServiceHandler
|
||||
handlerEx ServiceHandlerEx
|
||||
padding bool
|
||||
brutal BrutalOptions
|
||||
}
|
||||
|
|
@ -30,6 +39,7 @@ type ServiceOptions struct {
|
|||
NewStreamContext func(context.Context, net.Conn) context.Context
|
||||
Logger logger.ContextLogger
|
||||
Handler ServiceHandler
|
||||
HandlerEx ServiceHandlerEx
|
||||
Padding bool
|
||||
Brutal BrutalOptions
|
||||
}
|
||||
|
|
@ -42,12 +52,26 @@ func NewService(options ServiceOptions) (*Service, error) {
|
|||
newStreamContext: options.NewStreamContext,
|
||||
logger: options.Logger,
|
||||
handler: options.Handler,
|
||||
handlerEx: options.HandlerEx,
|
||||
padding: options.Padding,
|
||||
brutal: options.Brutal,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Deprecated: Use NewConnectionEx instead.
|
||||
func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
|
||||
return s.newConnection(ctx, conn, metadata.Source)
|
||||
}
|
||||
|
||||
func (s *Service) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
|
||||
err := s.newConnection(ctx, conn, source)
|
||||
N.CloseOnHandshakeFailure(conn, onClose, err)
|
||||
if err != nil {
|
||||
s.logger.ErrorContext(ctx, E.Cause(err, "process multiplex connection from ", source))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) newConnection(ctx context.Context, conn net.Conn, source M.Socksaddr) error {
|
||||
request, err := ReadRequest(conn)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -70,9 +94,10 @@ func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.M
|
|||
}
|
||||
streamCtx := s.newStreamContext(ctx, stream)
|
||||
go func() {
|
||||
hErr := s.newConnection(streamCtx, conn, stream, metadata)
|
||||
hErr := s.newSession(streamCtx, conn, stream, source)
|
||||
if hErr != nil {
|
||||
s.logger.ErrorContext(streamCtx, E.Cause(hErr, "handle connection"))
|
||||
stream.Close()
|
||||
s.logger.ErrorContext(streamCtx, E.Cause(hErr, "process multiplex stream"))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
@ -83,13 +108,13 @@ func (s *Service) NewConnection(ctx context.Context, conn net.Conn, metadata M.M
|
|||
return group.Run(ctx)
|
||||
}
|
||||
|
||||
func (s *Service) newConnection(ctx context.Context, sessionConn net.Conn, stream net.Conn, metadata M.Metadata) error {
|
||||
func (s *Service) newSession(ctx context.Context, sessionConn net.Conn, stream net.Conn, source M.Socksaddr) error {
|
||||
stream = &wrapStream{stream}
|
||||
request, err := ReadStreamRequest(stream)
|
||||
if err != nil {
|
||||
return E.Cause(err, "read multiplex stream request")
|
||||
}
|
||||
metadata.Destination = request.Destination
|
||||
destination := request.Destination
|
||||
if request.Network == N.NetworkTCP {
|
||||
conn := &serverConn{ExtendedConn: bufio.NewExtendedConn(stream)}
|
||||
if request.Destination.Fqdn == BrutalExchangeDomain {
|
||||
|
|
@ -127,20 +152,28 @@ func (s *Service) newConnection(ctx context.Context, sessionConn net.Conn, strea
|
|||
}
|
||||
return nil
|
||||
}
|
||||
s.logger.InfoContext(ctx, "inbound multiplex connection to ", metadata.Destination)
|
||||
s.handler.NewConnection(ctx, conn, metadata)
|
||||
stream.Close()
|
||||
s.logger.InfoContext(ctx, "inbound multiplex connection to ", destination)
|
||||
if s.handler != nil {
|
||||
//nolint:staticcheck
|
||||
s.handler.NewConnection(ctx, conn, M.Metadata{Source: source, Destination: destination})
|
||||
} else {
|
||||
s.handlerEx.NewConnectionEx(ctx, conn, source, destination, nil)
|
||||
}
|
||||
} else {
|
||||
var packetConn N.PacketConn
|
||||
if !request.PacketAddr {
|
||||
s.logger.InfoContext(ctx, "inbound multiplex packet connection to ", metadata.Destination)
|
||||
s.logger.InfoContext(ctx, "inbound multiplex packet connection to ", destination)
|
||||
packetConn = &serverPacketConn{ExtendedConn: bufio.NewExtendedConn(stream), destination: request.Destination}
|
||||
} else {
|
||||
s.logger.InfoContext(ctx, "inbound multiplex packet connection")
|
||||
packetConn = &serverPacketAddrConn{ExtendedConn: bufio.NewExtendedConn(stream)}
|
||||
}
|
||||
s.handler.NewPacketConnection(ctx, packetConn, metadata)
|
||||
stream.Close()
|
||||
if s.handler != nil {
|
||||
//nolint:staticcheck
|
||||
s.handler.NewPacketConnection(ctx, packetConn, M.Metadata{Source: source, Destination: destination})
|
||||
} else {
|
||||
s.handlerEx.NewPacketConnectionEx(ctx, packetConn, source, destination, nil)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import (
|
|||
"github.com/sagernet/sing/common/buf"
|
||||
M "github.com/sagernet/sing/common/metadata"
|
||||
N "github.com/sagernet/sing/common/network"
|
||||
"github.com/sagernet/sing/common/rw"
|
||||
"github.com/sagernet/sing/common/varbin"
|
||||
)
|
||||
|
||||
type serverConn struct {
|
||||
|
|
@ -24,11 +24,11 @@ func (c *serverConn) NeedHandshake() bool {
|
|||
|
||||
func (c *serverConn) HandshakeFailure(err error) error {
|
||||
errMessage := err.Error()
|
||||
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
|
||||
buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
|
||||
defer buffer.Release()
|
||||
common.Must(
|
||||
buffer.WriteByte(statusError),
|
||||
rw.WriteVString(buffer, errMessage),
|
||||
varbin.Write(buffer, binary.BigEndian, errMessage),
|
||||
)
|
||||
return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
|
||||
}
|
||||
|
|
@ -88,11 +88,11 @@ func (c *serverPacketConn) NeedHandshake() bool {
|
|||
|
||||
func (c *serverPacketConn) HandshakeFailure(err error) error {
|
||||
errMessage := err.Error()
|
||||
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
|
||||
buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
|
||||
defer buffer.Release()
|
||||
common.Must(
|
||||
buffer.WriteByte(statusError),
|
||||
rw.WriteVString(buffer, errMessage),
|
||||
varbin.Write(buffer, binary.BigEndian, errMessage),
|
||||
)
|
||||
return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
|
||||
}
|
||||
|
|
@ -188,11 +188,11 @@ func (c *serverPacketAddrConn) NeedHandshake() bool {
|
|||
|
||||
func (c *serverPacketAddrConn) HandshakeFailure(err error) error {
|
||||
errMessage := err.Error()
|
||||
buffer := buf.NewSize(1 + rw.UVariantLen(uint64(len(errMessage))) + len(errMessage))
|
||||
buffer := buf.NewSize(1 + varbin.UvarintLen(uint64(len(errMessage))) + len(errMessage))
|
||||
defer buffer.Release()
|
||||
common.Must(
|
||||
buffer.WriteByte(statusError),
|
||||
rw.WriteVString(buffer, errMessage),
|
||||
varbin.Write(buffer, binary.BigEndian, errMessage),
|
||||
)
|
||||
return common.Error(c.ExtendedConn.Write(buffer.Bytes()))
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue