From d92d36ddd9b93d8dea713d2802d0a94bfe974749 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Wed, 28 Apr 2021 23:29:42 +0100 Subject: [PATCH 1/5] VMess AEAD based packet length --- common/protocol/headers.go | 2 + infra/conf/vmess.go | 8 ++- proxy/vmess/account.go | 15 ++++- proxy/vmess/encoding/auth.go | 10 +++ proxy/vmess/encoding/client.go | 46 +++++++++++++ proxy/vmess/encoding/server.go | 46 +++++++++++++ proxy/vmess/outbound/outbound.go | 4 ++ testing/scenarios/vmess_test.go | 107 +++++++++++++++++++++++++++++++ 8 files changed, 232 insertions(+), 6 deletions(-) diff --git a/common/protocol/headers.go b/common/protocol/headers.go index 11bb641d96e..6ea8f08a7a7 100644 --- a/common/protocol/headers.go +++ b/common/protocol/headers.go @@ -38,6 +38,8 @@ const ( RequestOptionChunkMasking bitmask.Byte = 0x04 RequestOptionGlobalPadding bitmask.Byte = 0x08 + + RequestOptionAuthenticatedLength bitmask.Byte = 0x10 ) type RequestHeader struct { diff --git a/infra/conf/vmess.go b/infra/conf/vmess.go index 4c330d0d512..486dec93122 100644 --- a/infra/conf/vmess.go +++ b/infra/conf/vmess.go @@ -14,9 +14,10 @@ import ( ) type VMessAccount struct { - ID string `json:"id"` - AlterIds uint16 `json:"alterId"` - Security string `json:"security"` + ID string `json:"id"` + AlterIds uint16 `json:"alterId"` + Security string `json:"security"` + Experiments string `json:"experiments"` } // Build implements Buildable @@ -42,6 +43,7 @@ func (a *VMessAccount) Build() *vmess.Account { SecuritySettings: &protocol.SecurityConfig{ Type: st, }, + TestsEnabled: a.Experiments, } } diff --git a/proxy/vmess/account.go b/proxy/vmess/account.go index 6da196737af..cbd71eeff6b 100644 --- a/proxy/vmess/account.go +++ b/proxy/vmess/account.go @@ -3,6 +3,8 @@ package vmess import ( + "strings" + "github.com/v2fly/v2ray-core/v4/common/dice" "github.com/v2fly/v2ray-core/v4/common/protocol" "github.com/v2fly/v2ray-core/v4/common/uuid" @@ -16,6 +18,8 @@ type MemoryAccount struct { AlterIDs []*protocol.ID // Security type of the account. Used for client connections. Security protocol.SecurityType + + AuthenticatedLengthExperiment bool } // AnyValidID returns an ID that is either the main ID or one of the alternative IDs if any. @@ -43,9 +47,14 @@ func (a *Account) AsAccount() (protocol.Account, error) { return nil, newError("failed to parse ID").Base(err).AtError() } protoID := protocol.NewID(id) + var AuthenticatedLength bool + if strings.Contains(a.TestsEnabled, "AuthenticatedLength") { + AuthenticatedLength = true + } return &MemoryAccount{ - ID: protoID, - AlterIDs: protocol.NewAlterIDs(protoID, uint16(a.AlterId)), - Security: a.SecuritySettings.GetSecurityType(), + ID: protoID, + AlterIDs: protocol.NewAlterIDs(protoID, uint16(a.AlterId)), + Security: a.SecuritySettings.GetSecurityType(), + AuthenticatedLengthExperiment: AuthenticatedLength, }, nil } diff --git a/proxy/vmess/encoding/auth.go b/proxy/vmess/encoding/auth.go index d57a31be7e8..b3a070c0ca7 100644 --- a/proxy/vmess/encoding/auth.go +++ b/proxy/vmess/encoding/auth.go @@ -5,6 +5,8 @@ import ( "encoding/binary" "hash/fnv" + "github.com/v2fly/v2ray-core/v4/common/crypto" + "golang.org/x/crypto/sha3" "github.com/v2fly/v2ray-core/v4/common" @@ -117,3 +119,11 @@ func (s *ShakeSizeParser) NextPaddingLen() uint16 { func (s *ShakeSizeParser) MaxPaddingLen() uint16 { return 64 } + +type AEADSizeParser struct { + crypto.AEADChunkSizeParser +} + +func NewAEADSizeParser(auth *crypto.AEADAuthenticator) *AEADSizeParser { + return &AEADSizeParser{crypto.AEADChunkSizeParser{Auth: auth}} +} diff --git a/proxy/vmess/encoding/client.go b/proxy/vmess/encoding/client.go index 97daac6e490..24e7e8c2728 100644 --- a/proxy/vmess/encoding/client.go +++ b/proxy/vmess/encoding/client.go @@ -171,6 +171,17 @@ func (c *ClientSession) EncodeRequestBody(request *protocol.RequestHeader, write NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(c.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD := crypto.NewAesGcm(AuthenticatedLengthKey) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationWriter(auth, sizeParser, writer, request.Command.TransferType(), padding) case protocol.SecurityType_CHACHA20_POLY1305: aead, err := chacha20poly1305.New(GenerateChacha20Poly1305Key(c.requestBodyKey[:])) @@ -181,6 +192,18 @@ func (c *ClientSession) EncodeRequestBody(request *protocol.RequestHeader, write NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(c.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD, err := chacha20poly1305.New(GenerateChacha20Poly1305Key(AuthenticatedLengthKey)) + common.Must(err) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationWriter(auth, sizeParser, writer, request.Command.TransferType(), padding) default: panic("Unknown security type.") @@ -312,6 +335,17 @@ func (c *ClientSession) DecodeResponseBody(request *protocol.RequestHeader, read NonceGenerator: GenerateChunkNonce(c.responseBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(c.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD := crypto.NewAesGcm(AuthenticatedLengthKey) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationReader(auth, sizeParser, reader, request.Command.TransferType(), padding) case protocol.SecurityType_CHACHA20_POLY1305: aead, _ := chacha20poly1305.New(GenerateChacha20Poly1305Key(c.responseBodyKey[:])) @@ -321,6 +355,18 @@ func (c *ClientSession) DecodeResponseBody(request *protocol.RequestHeader, read NonceGenerator: GenerateChunkNonce(c.responseBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(c.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD, err := chacha20poly1305.New(GenerateChacha20Poly1305Key(AuthenticatedLengthKey)) + common.Must(err) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(c.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationReader(auth, sizeParser, reader, request.Command.TransferType(), padding) default: panic("Unknown security type.") diff --git a/proxy/vmess/encoding/server.go b/proxy/vmess/encoding/server.go index 9a3eb3e0c9a..de4ccbd617b 100644 --- a/proxy/vmess/encoding/server.go +++ b/proxy/vmess/encoding/server.go @@ -360,6 +360,17 @@ func (s *ServerSession) DecodeRequestBody(request *protocol.RequestHeader, reade NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(s.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD := crypto.NewAesGcm(AuthenticatedLengthKey) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationReader(auth, sizeParser, reader, request.Command.TransferType(), padding) case protocol.SecurityType_CHACHA20_POLY1305: @@ -370,6 +381,18 @@ func (s *ServerSession) DecodeRequestBody(request *protocol.RequestHeader, reade NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(s.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD, err := chacha20poly1305.New(GenerateChacha20Poly1305Key(AuthenticatedLengthKey)) + common.Must(err) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationReader(auth, sizeParser, reader, request.Command.TransferType(), padding) default: @@ -478,6 +501,17 @@ func (s *ServerSession) EncodeResponseBody(request *protocol.RequestHeader, writ NonceGenerator: GenerateChunkNonce(s.responseBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(s.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD := crypto.NewAesGcm(AuthenticatedLengthKey) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationWriter(auth, sizeParser, writer, request.Command.TransferType(), padding) case protocol.SecurityType_CHACHA20_POLY1305: @@ -488,6 +522,18 @@ func (s *ServerSession) EncodeResponseBody(request *protocol.RequestHeader, writ NonceGenerator: GenerateChunkNonce(s.responseBodyIV[:], uint32(aead.NonceSize())), AdditionalDataGenerator: crypto.GenerateEmptyBytes(), } + if request.Option.Has(protocol.RequestOptionAuthenticatedLength) { + AuthenticatedLengthKey := vmessaead.KDF16(s.requestBodyKey[:], "auth_len") + AuthenticatedLengthKeyAEAD, err := chacha20poly1305.New(GenerateChacha20Poly1305Key(AuthenticatedLengthKey)) + common.Must(err) + + lengthAuth := &crypto.AEADAuthenticator{ + AEAD: AuthenticatedLengthKeyAEAD, + NonceGenerator: GenerateChunkNonce(s.requestBodyIV[:], uint32(aead.NonceSize())), + AdditionalDataGenerator: crypto.GenerateEmptyBytes(), + } + sizeParser = NewAEADSizeParser(lengthAuth) + } return crypto.NewAuthenticationWriter(auth, sizeParser, writer, request.Command.TransferType(), padding) default: diff --git a/proxy/vmess/outbound/outbound.go b/proxy/vmess/outbound/outbound.go index c41f767cebc..9f7da6fe46f 100644 --- a/proxy/vmess/outbound/outbound.go +++ b/proxy/vmess/outbound/outbound.go @@ -116,6 +116,10 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte request.Option.Clear(protocol.RequestOptionChunkMasking) } + if account.AuthenticatedLengthExperiment { + request.Option.Set(protocol.RequestOptionAuthenticatedLength) + } + input := link.Reader output := link.Writer diff --git a/testing/scenarios/vmess_test.go b/testing/scenarios/vmess_test.go index 197c440bb5c..48b4dbc83a4 100644 --- a/testing/scenarios/vmess_test.go +++ b/testing/scenarios/vmess_test.go @@ -1287,3 +1287,110 @@ func TestVMessZero(t *testing.T) { t.Error(err) } } + +func TestVMessGCMLengthAuth(t *testing.T) { + tcpServer := tcp.Server{ + MsgProcessor: xor, + } + dest, err := tcpServer.Start() + common.Must(err) + defer tcpServer.Close() + + userID := protocol.NewID(uuid.New()) + serverPort := tcp.PickPort() + serverConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&log.Config{ + ErrorLogLevel: clog.Severity_Debug, + ErrorLogType: log.LogType_Console, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(serverPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&inbound.Config{ + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + }), + }, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&freedom.Config{}), + }, + }, + } + + clientPort := tcp.PickPort() + clientConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&log.Config{ + ErrorLogLevel: clog.Severity_Debug, + ErrorLogType: log.LogType_Console, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(clientPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&dokodemo.Config{ + Address: net.NewIPOrDomain(dest.Address), + Port: uint32(dest.Port), + NetworkList: &net.NetworkList{ + Network: []net.Network{net.Network_TCP}, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&outbound.Config{ + Receiver: []*protocol.ServerEndpoint{ + { + Address: net.NewIPOrDomain(net.LocalHostIP), + Port: uint32(serverPort), + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + SecuritySettings: &protocol.SecurityConfig{ + Type: protocol.SecurityType_AES128_GCM, + }, + TestsEnabled: "AuthenticatedLength", + }), + }, + }, + }, + }, + }), + }, + }, + } + + servers, err := InitializeServerConfigs(serverConfig, clientConfig) + if err != nil { + t.Fatal("Failed to initialize all servers: ", err.Error()) + } + defer CloseAllServers(servers) + + var errg errgroup.Group + for i := 0; i < 10; i++ { + errg.Go(testTCPConn(clientPort, 10240*1024, time.Second*40)) + } + + if err := errg.Wait(); err != nil { + t.Error(err) + } +} From d05b19dfc95c5d41883ac83c8c618b92e51f12a8 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Thu, 29 Apr 2021 14:28:52 +0100 Subject: [PATCH 2/5] Added experiment to avoid sending the termination signal --- proxy/vmess/account.go | 7 ++++++- proxy/vmess/inbound/inbound.go | 4 +++- proxy/vmess/outbound/outbound.go | 2 +- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/proxy/vmess/account.go b/proxy/vmess/account.go index cbd71eeff6b..5023684dcb6 100644 --- a/proxy/vmess/account.go +++ b/proxy/vmess/account.go @@ -20,6 +20,7 @@ type MemoryAccount struct { Security protocol.SecurityType AuthenticatedLengthExperiment bool + NoTerminationSignal bool } // AnyValidID returns an ID that is either the main ID or one of the alternative IDs if any. @@ -47,14 +48,18 @@ func (a *Account) AsAccount() (protocol.Account, error) { return nil, newError("failed to parse ID").Base(err).AtError() } protoID := protocol.NewID(id) - var AuthenticatedLength bool + var AuthenticatedLength, NoTerminationSignal bool if strings.Contains(a.TestsEnabled, "AuthenticatedLength") { AuthenticatedLength = true } + if strings.Contains(a.TestsEnabled, "NoTerminationSignal") { + NoTerminationSignal = true + } return &MemoryAccount{ ID: protoID, AlterIDs: protocol.NewAlterIDs(protoID, uint16(a.AlterId)), Security: a.SecuritySettings.GetSecurityType(), AuthenticatedLengthExperiment: AuthenticatedLength, + NoTerminationSignal: NoTerminationSignal, }, nil } diff --git a/proxy/vmess/inbound/inbound.go b/proxy/vmess/inbound/inbound.go index 8883f9d80a7..e4c56c5688c 100644 --- a/proxy/vmess/inbound/inbound.go +++ b/proxy/vmess/inbound/inbound.go @@ -203,7 +203,9 @@ func transferResponse(timer signal.ActivityUpdater, session *encoding.ServerSess return err } - if request.Option.Has(protocol.RequestOptionChunkStream) { + account := request.User.Account.(*vmess.MemoryAccount) + + if request.Option.Has(protocol.RequestOptionChunkStream) && !account.NoTerminationSignal { if err := bodyWriter.WriteMultiBuffer(buf.MultiBuffer{}); err != nil { return err } diff --git a/proxy/vmess/outbound/outbound.go b/proxy/vmess/outbound/outbound.go index 9f7da6fe46f..8fe3d6a0551 100644 --- a/proxy/vmess/outbound/outbound.go +++ b/proxy/vmess/outbound/outbound.go @@ -155,7 +155,7 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte return err } - if request.Option.Has(protocol.RequestOptionChunkStream) { + if request.Option.Has(protocol.RequestOptionChunkStream) && !account.NoTerminationSignal { if err := bodyWriter.WriteMultiBuffer(buf.MultiBuffer{}); err != nil { return err } From 1b52b52cf7d737ae5198dc4a3862dc77a82c7177 Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Thu, 29 Apr 2021 14:41:50 +0100 Subject: [PATCH 3/5] Added test for no terminate signal --- testing/scenarios/vmess_test.go | 108 ++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/testing/scenarios/vmess_test.go b/testing/scenarios/vmess_test.go index 48b4dbc83a4..96db0940693 100644 --- a/testing/scenarios/vmess_test.go +++ b/testing/scenarios/vmess_test.go @@ -1394,3 +1394,111 @@ func TestVMessGCMLengthAuth(t *testing.T) { t.Error(err) } } + +func TestVMessGCMLengthAuthPlusNoTerminationSignal(t *testing.T) { + tcpServer := tcp.Server{ + MsgProcessor: xor, + } + dest, err := tcpServer.Start() + common.Must(err) + defer tcpServer.Close() + + userID := protocol.NewID(uuid.New()) + serverPort := tcp.PickPort() + serverConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&log.Config{ + ErrorLogLevel: clog.Severity_Debug, + ErrorLogType: log.LogType_Console, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(serverPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&inbound.Config{ + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + TestsEnabled: "AuthenticatedLength|NoTerminationSignal", + }), + }, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&freedom.Config{}), + }, + }, + } + + clientPort := tcp.PickPort() + clientConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&log.Config{ + ErrorLogLevel: clog.Severity_Debug, + ErrorLogType: log.LogType_Console, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(clientPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&dokodemo.Config{ + Address: net.NewIPOrDomain(dest.Address), + Port: uint32(dest.Port), + NetworkList: &net.NetworkList{ + Network: []net.Network{net.Network_TCP}, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&outbound.Config{ + Receiver: []*protocol.ServerEndpoint{ + { + Address: net.NewIPOrDomain(net.LocalHostIP), + Port: uint32(serverPort), + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + SecuritySettings: &protocol.SecurityConfig{ + Type: protocol.SecurityType_AES128_GCM, + }, + TestsEnabled: "AuthenticatedLength|NoTerminationSignal", + }), + }, + }, + }, + }, + }), + }, + }, + } + + servers, err := InitializeServerConfigs(serverConfig, clientConfig) + if err != nil { + t.Fatal("Failed to initialize all servers: ", err.Error()) + } + defer CloseAllServers(servers) + + var errg errgroup.Group + for i := 0; i < 10; i++ { + errg.Go(testTCPConn(clientPort, 10240*1024, time.Second*40)) + } + + if err := errg.Wait(); err != nil { + t.Error(err) + } +} From f4c0d46b2b5350da6d52fb5fa91ce4ed5374b60a Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Fri, 30 Apr 2021 12:08:15 +0100 Subject: [PATCH 4/5] unified drain support for vmess and shadowsockets --- common/drain/drain.go | 10 +++++ common/drain/drainer.go | 62 ++++++++++++++++++++++++++ proxy/shadowsocks/protocol.go | 63 ++++++++++++++------------- proxy/vmess/encoding/client.go | 32 +++++++++++--- proxy/vmess/encoding/encoding_test.go | 6 +-- proxy/vmess/encoding/server.go | 36 +++++---------- proxy/vmess/outbound/outbound.go | 10 ++++- 7 files changed, 153 insertions(+), 66 deletions(-) create mode 100644 common/drain/drain.go create mode 100644 common/drain/drainer.go diff --git a/common/drain/drain.go b/common/drain/drain.go new file mode 100644 index 00000000000..d48d067aec2 --- /dev/null +++ b/common/drain/drain.go @@ -0,0 +1,10 @@ +package drain + +import "io" + +//go:generate go run github.com/v2fly/v2ray-core/v4/common/errors/errorgen + +type Drainer interface { + AcknowledgeReceive(size int) + Drain(reader io.Reader) error +} diff --git a/common/drain/drainer.go b/common/drain/drainer.go new file mode 100644 index 00000000000..5ed887abc6b --- /dev/null +++ b/common/drain/drainer.go @@ -0,0 +1,62 @@ +package drain + +import ( + "io" + "io/ioutil" + + "github.com/v2fly/v2ray-core/v4/common/dice" +) + +type BehaviorSeedLimitedDrainer struct { + DrainSize int +} + +func NewBehaviorSeedLimitedDrainer(behaviorSeed int64, drainFoundation, maxBaseDrainSize, maxRandDrain int) (Drainer, error) { + behaviorRand := dice.NewDeterministicDice(behaviorSeed) + BaseDrainSize := behaviorRand.Roll(maxBaseDrainSize) + RandDrainMax := behaviorRand.Roll(maxRandDrain) + 1 + RandDrainRolled := dice.Roll(RandDrainMax) + DrainSize := drainFoundation + BaseDrainSize + RandDrainRolled + return &BehaviorSeedLimitedDrainer{DrainSize: DrainSize}, nil +} + +func (d *BehaviorSeedLimitedDrainer) AcknowledgeReceive(size int) { + d.DrainSize -= size +} + +func (d *BehaviorSeedLimitedDrainer) Drain(reader io.Reader) error { + if d.DrainSize > 0 { + err := drainReadN(reader, d.DrainSize) + if err == nil { + return newError("drained connection") + } + return newError("unable to drain connection").Base(err) + } + return nil +} + +func drainReadN(reader io.Reader, n int) error { + _, err := io.CopyN(ioutil.Discard, reader, int64(n)) + return err +} + +func WithError(drainer Drainer, reader io.Reader, err error) error { + drainErr := drainer.Drain(reader) + if drainErr == nil { + return err + } + return newError(drainErr).Base(err) +} + +type NopDrainer struct{} + +func (n NopDrainer) AcknowledgeReceive(size int) { +} + +func (n NopDrainer) Drain(reader io.Reader) error { + return nil +} + +func NewNopDrainer() Drainer { + return &NopDrainer{} +} diff --git a/proxy/shadowsocks/protocol.go b/proxy/shadowsocks/protocol.go index 6f059f569b3..3654e5b4de1 100644 --- a/proxy/shadowsocks/protocol.go +++ b/proxy/shadowsocks/protocol.go @@ -8,11 +8,10 @@ import ( "crypto/sha256" "hash/crc32" "io" - "io/ioutil" "github.com/v2fly/v2ray-core/v4/common" "github.com/v2fly/v2ray-core/v4/common/buf" - "github.com/v2fly/v2ray-core/v4/common/dice" + "github.com/v2fly/v2ray-core/v4/common/drain" "github.com/v2fly/v2ray-core/v4/common/net" "github.com/v2fly/v2ray-core/v4/common/protocol" ) @@ -39,12 +38,11 @@ func ReadTCPSession(user *protocol.MemoryUser, reader io.Reader) (*protocol.Requ behaviorSeed := crc32.ChecksumIEEE(hashkdf.Sum(nil)) - behaviorRand := dice.NewDeterministicDice(int64(behaviorSeed)) - BaseDrainSize := behaviorRand.Roll(3266) - RandDrainMax := behaviorRand.Roll(64) + 1 - RandDrainRolled := dice.Roll(RandDrainMax) - DrainSize := BaseDrainSize + 16 + 38 + RandDrainRolled - readSizeRemain := DrainSize + drainer, err := drain.NewBehaviorSeedLimitedDrainer(int64(behaviorSeed), 16+38, 3266, 64) + + if err != nil { + return nil, nil, newError("failed to initialize drainer").Base(err) + } buffer := buf.New() defer buffer.Release() @@ -53,9 +51,8 @@ func ReadTCPSession(user *protocol.MemoryUser, reader io.Reader) (*protocol.Requ var iv []byte if ivLen > 0 { if _, err := buffer.ReadFullFrom(reader, ivLen); err != nil { - readSizeRemain -= int(buffer.Len()) - DrainConnN(reader, readSizeRemain) - return nil, nil, newError("failed to read IV").Base(err) + drainer.AcknowledgeReceive(int(buffer.Len())) + return nil, nil, drain.WithError(drainer, reader, newError("failed to read IV").Base(err)) } iv = append([]byte(nil), buffer.BytesTo(ivLen)...) @@ -63,9 +60,8 @@ func ReadTCPSession(user *protocol.MemoryUser, reader io.Reader) (*protocol.Requ r, err := account.Cipher.NewDecryptionReader(account.Key, iv, reader) if err != nil { - readSizeRemain -= int(buffer.Len()) - DrainConnN(reader, readSizeRemain) - return nil, nil, newError("failed to initialize decoding stream").Base(err).AtError() + drainer.AcknowledgeReceive(int(buffer.Len())) + return nil, nil, drain.WithError(drainer, reader, newError("failed to initialize decoding stream").Base(err).AtError()) } br := &buf.BufferedReader{Reader: r} @@ -75,39 +71,31 @@ func ReadTCPSession(user *protocol.MemoryUser, reader io.Reader) (*protocol.Requ Command: protocol.RequestCommandTCP, } - readSizeRemain -= int(buffer.Len()) + drainer.AcknowledgeReceive(int(buffer.Len())) buffer.Clear() addr, port, err := addrParser.ReadAddressPort(buffer, br) if err != nil { - readSizeRemain -= int(buffer.Len()) - DrainConnN(reader, readSizeRemain) - return nil, nil, newError("failed to read address").Base(err) + drainer.AcknowledgeReceive(int(buffer.Len())) + return nil, nil, drain.WithError(drainer, reader, newError("failed to read address").Base(err)) } request.Address = addr request.Port = port if request.Address == nil { - readSizeRemain -= int(buffer.Len()) - DrainConnN(reader, readSizeRemain) - return nil, nil, newError("invalid remote address.") + drainer.AcknowledgeReceive(int(buffer.Len())) + return nil, nil, drain.WithError(drainer, reader, newError("invalid remote address.")) } if ivError := account.CheckIV(iv); ivError != nil { - readSizeRemain -= int(buffer.Len()) - DrainConnN(reader, readSizeRemain) - return nil, nil, newError("failed iv check").Base(ivError) + drainer.AcknowledgeReceive(int(buffer.Len())) + return nil, nil, drain.WithError(drainer, reader, newError("failed iv check").Base(ivError)) } return request, br, nil } -func DrainConnN(reader io.Reader, n int) error { - _, err := io.CopyN(ioutil.Discard, reader, int64(n)) - return err -} - // WriteTCPRequest writes Shadowsocks request into the given writer, and returns a writer for body. func WriteTCPRequest(request *protocol.RequestHeader, writer io.Writer) (buf.Writer, error) { user := request.User @@ -146,16 +134,29 @@ func WriteTCPRequest(request *protocol.RequestHeader, writer io.Writer) (buf.Wri func ReadTCPResponse(user *protocol.MemoryUser, reader io.Reader) (buf.Reader, error) { account := user.Account.(*MemoryAccount) + hashkdf := hmac.New(sha256.New, []byte("SSBSKDF")) + hashkdf.Write(account.Key) + + behaviorSeed := crc32.ChecksumIEEE(hashkdf.Sum(nil)) + + drainer, err := drain.NewBehaviorSeedLimitedDrainer(int64(behaviorSeed), 16+38, 3266, 64) + + if err != nil { + return nil, newError("failed to initialize drainer").Base(err) + } + var iv []byte if account.Cipher.IVSize() > 0 { iv = make([]byte, account.Cipher.IVSize()) - if _, err := io.ReadFull(reader, iv); err != nil { + if n, err := io.ReadFull(reader, iv); err != nil { return nil, newError("failed to read IV").Base(err) + } else { // nolint: golint + drainer.AcknowledgeReceive(n) } } if ivError := account.CheckIV(iv); ivError != nil { - return nil, newError("failed iv check").Base(ivError) + return nil, drain.WithError(drainer, reader, newError("failed iv check").Base(ivError)) } return account.Cipher.NewDecryptionReader(account.Key, iv, reader) diff --git a/proxy/vmess/encoding/client.go b/proxy/vmess/encoding/client.go index 24e7e8c2728..73ca6037d34 100644 --- a/proxy/vmess/encoding/client.go +++ b/proxy/vmess/encoding/client.go @@ -13,6 +13,8 @@ import ( "hash/fnv" "io" + "github.com/v2fly/v2ray-core/v4/common/drain" + "golang.org/x/crypto/chacha20poly1305" "github.com/v2fly/v2ray-core/v4/common" @@ -44,10 +46,12 @@ type ClientSession struct { responseBodyIV [16]byte responseReader io.Reader responseHeader byte + + readDrainer drain.Drainer } // NewClientSession creates a new ClientSession. -func NewClientSession(ctx context.Context, isAEAD bool, idHash protocol.IDHash) *ClientSession { +func NewClientSession(ctx context.Context, isAEAD bool, idHash protocol.IDHash, behaviorSeed int64) *ClientSession { session := &ClientSession{ isAEAD: isAEAD, idHash: idHash, @@ -68,6 +72,14 @@ func NewClientSession(ctx context.Context, isAEAD bool, idHash protocol.IDHash) BodyIV := sha256.Sum256(session.requestBodyIV[:]) copy(session.responseBodyIV[:], BodyIV[:16]) } + { + var err error + session.readDrainer, err = drain.NewBehaviorSeedLimitedDrainer(behaviorSeed, 18, 3266, 64) + if err != nil { + newError("unable to initialize drainer").Base(err).WriteToLog() + session.readDrainer = drain.NewNopDrainer() + } + } return session } @@ -225,11 +237,14 @@ func (c *ClientSession) DecodeResponseHeader(reader io.Reader) (*protocol.Respon var decryptedResponseHeaderLength int var decryptedResponseHeaderLengthBinaryDeserializeBuffer uint16 - if _, err := io.ReadFull(reader, aeadEncryptedResponseHeaderLength[:]); err != nil { - return nil, newError("Unable to Read Header Len").Base(err) + if n, err := io.ReadFull(reader, aeadEncryptedResponseHeaderLength[:]); err != nil { + c.readDrainer.AcknowledgeReceive(n) + return nil, drain.WithError(c.readDrainer, reader, newError("Unable to Read Header Len").Base(err)) + } else { // nolint: golint + c.readDrainer.AcknowledgeReceive(n) } if decryptedResponseHeaderLengthBinaryBuffer, err := aeadResponseHeaderLengthEncryptionAEAD.Open(nil, aeadResponseHeaderLengthEncryptionIV, aeadEncryptedResponseHeaderLength[:], nil); err != nil { - return nil, newError("Failed To Decrypt Length").Base(err) + return nil, drain.WithError(c.readDrainer, reader, newError("Failed To Decrypt Length").Base(err)) } else { // nolint: golint common.Must(binary.Read(bytes.NewReader(decryptedResponseHeaderLengthBinaryBuffer), binary.BigEndian, &decryptedResponseHeaderLengthBinaryDeserializeBuffer)) decryptedResponseHeaderLength = int(decryptedResponseHeaderLengthBinaryDeserializeBuffer) @@ -243,12 +258,15 @@ func (c *ClientSession) DecodeResponseHeader(reader io.Reader) (*protocol.Respon encryptedResponseHeaderBuffer := make([]byte, decryptedResponseHeaderLength+16) - if _, err := io.ReadFull(reader, encryptedResponseHeaderBuffer); err != nil { - return nil, newError("Unable to Read Header Data").Base(err) + if n, err := io.ReadFull(reader, encryptedResponseHeaderBuffer); err != nil { + c.readDrainer.AcknowledgeReceive(n) + return nil, drain.WithError(c.readDrainer, reader, newError("Unable to Read Header Data").Base(err)) + } else { // nolint: golint + c.readDrainer.AcknowledgeReceive(n) } if decryptedResponseHeaderBuffer, err := aeadResponseHeaderPayloadEncryptionAEAD.Open(nil, aeadResponseHeaderPayloadEncryptionIV, encryptedResponseHeaderBuffer, nil); err != nil { - return nil, newError("Failed To Decrypt Payload").Base(err) + return nil, drain.WithError(c.readDrainer, reader, newError("Failed To Decrypt Payload").Base(err)) } else { // nolint: golint c.responseReader = bytes.NewReader(decryptedResponseHeaderBuffer) } diff --git a/proxy/vmess/encoding/encoding_test.go b/proxy/vmess/encoding/encoding_test.go index 71579b0d066..e2d2ea90c3e 100644 --- a/proxy/vmess/encoding/encoding_test.go +++ b/proxy/vmess/encoding/encoding_test.go @@ -43,7 +43,7 @@ func TestRequestSerialization(t *testing.T) { } buffer := buf.New() - client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash) + client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash, 0) common.Must(client.EncodeRequestHeader(expectedRequest, buffer)) buffer2 := buf.New() @@ -93,7 +93,7 @@ func TestInvalidRequest(t *testing.T) { } buffer := buf.New() - client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash) + client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash, 0) common.Must(client.EncodeRequestHeader(expectedRequest, buffer)) buffer2 := buf.New() @@ -134,7 +134,7 @@ func TestMuxRequest(t *testing.T) { } buffer := buf.New() - client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash) + client := NewClientSession(context.TODO(), true, protocol.DefaultIDHash, 0) common.Must(client.EncodeRequestHeader(expectedRequest, buffer)) buffer2 := buf.New() diff --git a/proxy/vmess/encoding/server.go b/proxy/vmess/encoding/server.go index de4ccbd617b..ac0ca6acff3 100644 --- a/proxy/vmess/encoding/server.go +++ b/proxy/vmess/encoding/server.go @@ -9,17 +9,17 @@ import ( "encoding/binary" "hash/fnv" "io" - "io/ioutil" "sync" "time" + "github.com/v2fly/v2ray-core/v4/common/drain" + "golang.org/x/crypto/chacha20poly1305" "github.com/v2fly/v2ray-core/v4/common" "github.com/v2fly/v2ray-core/v4/common/bitmask" "github.com/v2fly/v2ray-core/v4/common/buf" "github.com/v2fly/v2ray-core/v4/common/crypto" - "github.com/v2fly/v2ray-core/v4/common/dice" "github.com/v2fly/v2ray-core/v4/common/net" "github.com/v2fly/v2ray-core/v4/common/protocol" "github.com/v2fly/v2ray-core/v4/common/task" @@ -139,24 +139,17 @@ func parseSecurityType(b byte) protocol.SecurityType { // DecodeRequestHeader decodes and returns (if successful) a RequestHeader from an input stream. func (s *ServerSession) DecodeRequestHeader(reader io.Reader) (*protocol.RequestHeader, error) { buffer := buf.New() - behaviorRand := dice.NewDeterministicDice(int64(s.userValidator.GetBehaviorSeed())) - BaseDrainSize := behaviorRand.Roll(3266) - RandDrainMax := behaviorRand.Roll(64) + 1 - RandDrainRolled := dice.Roll(RandDrainMax) - DrainSize := BaseDrainSize + 16 + 38 + RandDrainRolled - readSizeRemain := DrainSize + + drainer, err := drain.NewBehaviorSeedLimitedDrainer(int64(s.userValidator.GetBehaviorSeed()), 16+38, 3266, 64) + + if err != nil { + return nil, newError("failed to initialize drainer").Base(err) + } drainConnection := func(e error) error { // We read a deterministic generated length of data before closing the connection to offset padding read pattern - readSizeRemain -= int(buffer.Len()) - if readSizeRemain > 0 { - err := s.DrainConnN(reader, readSizeRemain) - if err != nil { - return newError("failed to drain connection DrainSize = ", BaseDrainSize, " ", RandDrainMax, " ", RandDrainRolled).Base(err).Base(e) - } - return newError("connection drained DrainSize = ", BaseDrainSize, " ", RandDrainMax, " ", RandDrainRolled).Base(e) - } - return e + drainer.AcknowledgeReceive(int(buffer.Len())) + return drain.WithError(drainer, reader, e) } defer func() { @@ -183,7 +176,7 @@ func (s *ServerSession) DecodeRequestHeader(reader io.Reader) (*protocol.Request aeadData, shouldDrain, bytesRead, errorReason := vmessaead.OpenVMessAEADHeader(fixedSizeCmdKey, fixedSizeAuthID, reader) if errorReason != nil { if shouldDrain { - readSizeRemain -= bytesRead + drainer.AcknowledgeReceive(bytesRead) return nil, drainConnection(newError("AEAD read failed").Base(errorReason)) } return nil, drainConnection(newError("AEAD read failed, drain skipped").Base(errorReason)) @@ -213,7 +206,7 @@ func (s *ServerSession) DecodeRequestHeader(reader io.Reader) (*protocol.Request return nil, drainConnection(newError("invalid user").Base(errorAEAD)) } - readSizeRemain -= int(buffer.Len()) + drainer.AcknowledgeReceive(int(buffer.Len())) buffer.Clear() if _, err := buffer.ReadFullFrom(decryptor, 38); err != nil { return nil, newError("failed to read request header").Base(err) @@ -540,8 +533,3 @@ func (s *ServerSession) EncodeResponseBody(request *protocol.RequestHeader, writ panic("Unknown security type.") } } - -func (s *ServerSession) DrainConnN(reader io.Reader, n int) error { - _, err := io.CopyN(ioutil.Discard, reader, int64(n)) - return err -} diff --git a/proxy/vmess/outbound/outbound.go b/proxy/vmess/outbound/outbound.go index 8fe3d6a0551..bc415d8b93b 100644 --- a/proxy/vmess/outbound/outbound.go +++ b/proxy/vmess/outbound/outbound.go @@ -6,6 +6,9 @@ package outbound import ( "context" + "crypto/hmac" + "crypto/sha256" + "hash/crc64" "time" core "github.com/v2fly/v2ray-core/v4" @@ -128,7 +131,12 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte isAEAD = true } - session := encoding.NewClientSession(ctx, isAEAD, protocol.DefaultIDHash) + hashkdf := hmac.New(sha256.New, []byte("VMessBF")) + hashkdf.Write(account.ID.Bytes()) + + behaviorSeed := crc64.Checksum(hashkdf.Sum(nil), crc64.MakeTable(crc64.ISO)) + + session := encoding.NewClientSession(ctx, isAEAD, protocol.DefaultIDHash, int64(behaviorSeed)) sessionPolicy := h.policyManager.ForLevel(request.User.Level) ctx, cancel := context.WithCancel(ctx) From 95e0395e6bde2708bc8e08d7ff9abfbad45ac92a Mon Sep 17 00:00:00 2001 From: Shelikhoo Date: Fri, 30 Apr 2021 12:27:28 +0100 Subject: [PATCH 5/5] drain: add generated file --- common/drain/errors.generated.go | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 common/drain/errors.generated.go diff --git a/common/drain/errors.generated.go b/common/drain/errors.generated.go new file mode 100644 index 00000000000..9406d35f687 --- /dev/null +++ b/common/drain/errors.generated.go @@ -0,0 +1,9 @@ +package drain + +import "github.com/v2fly/v2ray-core/v4/common/errors" + +type errPathObjHolder struct{} + +func newError(values ...interface{}) *errors.Error { + return errors.New(values...).WithPathObj(errPathObjHolder{}) +}