prototype of mtproto proxy

common/crypto/aes.go

@@ -10,15 +10,21 @@ import (
 // NewAesDecryptionStream creates a new AES encryption stream based on given key and IV.
 // Caller must ensure the length of key and IV is either 16, 24 or 32 bytes.
 func NewAesDecryptionStream(key []byte, iv []byte) cipher.Stream {
+	return NewAesStreamMethod(key, iv, cipher.NewCFBDecrypter)
+}
+
+func NewAesStreamMethod(key []byte, iv []byte, f func(cipher.Block, []byte) cipher.Stream) cipher.Stream {
 	aesBlock, err := aes.NewCipher(key)
 	common.Must(err)
-	return cipher.NewCFBDecrypter(aesBlock, iv)
+	return f(aesBlock, iv)
 }
 
 // NewAesEncryptionStream creates a new AES description stream based on given key and IV.
 // Caller must ensure the length of key and IV is either 16, 24 or 32 bytes.
 func NewAesEncryptionStream(key []byte, iv []byte) cipher.Stream {
-	aesBlock, err := aes.NewCipher(key)
-	common.Must(err)
-	return cipher.NewCFBEncrypter(aesBlock, iv)
+	return NewAesStreamMethod(key, iv, cipher.NewCFBEncrypter)
+}
+
+func NewAesCTRStream(key []byte, iv []byte) cipher.Stream {
+	return NewAesStreamMethod(key, iv, cipher.NewCTR)
 }

main/distro/all/all.go

@@ -26,6 +26,7 @@ import (
 	_ "v2ray.com/core/proxy/dokodemo"
 	_ "v2ray.com/core/proxy/freedom"
 	_ "v2ray.com/core/proxy/http"
+	_ "v2ray.com/core/proxy/mtproto"
 	_ "v2ray.com/core/proxy/shadowsocks"
 	_ "v2ray.com/core/proxy/socks"
 	_ "v2ray.com/core/proxy/vmess/inbound"

proxy/mtproto/auth.go

@@ -0,0 +1,108 @@
+package mtproto
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"io"
+	"sync"
+
+	"v2ray.com/core/common"
+)
+
+const (
+	HeaderSize = 64
+)
+
+type Authentication struct {
+	Header        [HeaderSize]byte
+	DecodingKey   [32]byte
+	EncodingKey   [32]byte
+	DecodingNonce [16]byte
+	EncodingNonce [16]byte
+}
+
+func (a *Authentication) DataCenterID() uint16 {
+	return (uint16(a.Header[61]) << 8) | uint16(a.Header[60])
+}
+
+func (a *Authentication) ApplySecret(b []byte) {
+	a.DecodingKey = sha256.Sum256(append(a.DecodingKey[:], b...))
+	a.EncodingKey = sha256.Sum256(append(a.EncodingKey[:], b...))
+}
+
+func generateRandomBytes(random []byte) {
+	for {
+		common.Must2(rand.Read(random[:]))
+
+		if random[0] == 0xef {
+			continue
+		}
+
+		val := (uint32(random[3]) << 24) | (uint32(random[2]) << 16) | (uint32(random[1]) << 8) | uint32(random[0])
+		if val == 0x44414548 || val == 0x54534f50 || val == 0x20544547 || val == 0x4954504f || val == 0xeeeeeeee {
+			continue
+		}
+
+		if 0x00000000 == (uint32(random[7])<<24)|(uint32(random[6])<<16)|(uint32(random[5])<<8)|uint32(random[4]) {
+			continue
+		}
+
+		return
+	}
+}
+
+func NewAuthentication() *Authentication {
+	auth := getAuthenticationObject()
+	random := auth.Header[:]
+	generateRandomBytes(random)
+	copy(auth.EncodingKey[:], random[8:])
+	copy(auth.EncodingNonce[:], random[8+32:])
+	keyivInverse := Inverse(random[8 : 8+32+16])
+	copy(auth.DecodingKey[:], keyivInverse)
+	copy(auth.DecodingNonce[:], keyivInverse[32:])
+	return auth
+}
+
+func ReadAuthentication(reader io.Reader) (*Authentication, error) {
+	auth := getAuthenticationObject()
+
+	if _, err := io.ReadFull(reader, auth.Header[:]); err != nil {
+		putAuthenticationObject(auth)
+		return nil, err
+	}
+
+	copy(auth.DecodingKey[:], auth.Header[8:])
+	copy(auth.DecodingNonce[:], auth.Header[8+32:])
+	keyivInverse := Inverse(auth.Header[8 : 8+32+16])
+	copy(auth.EncodingKey[:], keyivInverse)
+	copy(auth.EncodingNonce[:], keyivInverse[32:])
+
+	return auth, nil
+}
+
+// Inverse returns a new byte array. It is a sequence of bytes when the input is read from end to beginning.Inverse
+// Visible for testing only.
+func Inverse(b []byte) []byte {
+	lenb := len(b)
+	b2 := make([]byte, lenb)
+	for i, v := range b {
+		b2[lenb-i-1] = v
+	}
+	return b2
+}
+
+var (
+	authPool = sync.Pool{
+		New: func() interface{} {
+			return new(Authentication)
+		},
+	}
+)
+
+func getAuthenticationObject() *Authentication {
+	return authPool.Get().(*Authentication)
+}
+
+func putAuthenticationObject(auth *Authentication) {
+	authPool.Put(auth)
+}

proxy/mtproto/auth_test.go

@@ -0,0 +1,23 @@
+package mtproto_test
+
+import (
+	"crypto/rand"
+	"testing"
+
+	"v2ray.com/core/common"
+	. "v2ray.com/core/proxy/mtproto"
+	. "v2ray.com/ext/assert"
+)
+
+func TestInverse(t *testing.T) {
+	assert := With(t)
+
+	b := make([]byte, 64)
+	common.Must2(rand.Read(b))
+
+	bi := Inverse(b)
+	assert(b[0], NotEquals, bi[0])
+
+	bii := Inverse(bi)
+	assert(bii, Equals, b)
+}

proxy/mtproto/client.go

@@ -0,0 +1,75 @@
+package mtproto
+
+import (
+	"context"
+
+	"v2ray.com/core"
+	"v2ray.com/core/common"
+	"v2ray.com/core/common/buf"
+	"v2ray.com/core/common/crypto"
+	"v2ray.com/core/common/net"
+	"v2ray.com/core/common/task"
+	"v2ray.com/core/proxy"
+)
+
+type Client struct {
+}
+
+func NewClient(ctx context.Context, config *ClientConfig) (*Client, error) {
+	return &Client{}, nil
+}
+
+func (c *Client) Process(ctx context.Context, link *core.Link, dialer proxy.Dialer) error {
+	dest, ok := proxy.TargetFromContext(ctx)
+	if !ok {
+		return newError("unknown destination.")
+	}
+
+	if dest.Network != net.Network_TCP {
+		return newError("not TCP traffic", dest)
+	}
+
+	conn, err := dialer.Dial(ctx, dest)
+	if err != nil {
+		return newError("failed to dial to ", dest).Base(err).AtWarning()
+	}
+	defer conn.Close() // nolint: errcheck
+
+	auth := NewAuthentication()
+	defer putAuthenticationObject(auth)
+
+	request := func() error {
+		encryptor := crypto.NewAesCTRStream(auth.EncodingKey[:], auth.EncodingNonce[:])
+
+		var header [HeaderSize]byte
+		encryptor.XORKeyStream(header[:], auth.Header[:])
+		copy(header[:56], auth.Header[:])
+
+		if _, err := conn.Write(header[:]); err != nil {
+			return newError("failed to write auth header").Base(err)
+		}
+
+		connWriter := buf.NewWriter(crypto.NewCryptionWriter(encryptor, conn))
+		return buf.Copy(link.Reader, connWriter)
+	}
+
+	response := func() error {
+		decryptor := crypto.NewAesCTRStream(auth.DecodingKey[:], auth.DecodingNonce[:])
+
+		connReader := buf.NewReader(crypto.NewCryptionReader(decryptor, conn))
+		return buf.Copy(connReader, link.Writer)
+	}
+
+	var responseDoneAndCloseWriter = task.Single(response, task.OnSuccess(task.Close(link.Writer)))
+	if err := task.Run(task.WithContext(ctx), task.Parallel(request, responseDoneAndCloseWriter))(); err != nil {
+		return newError("connection ends").Base(err)
+	}
+
+	return nil
+}
+
+func init() {
+	common.Must(common.RegisterConfig((*ClientConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
+		return NewClient(ctx, config.(*ClientConfig))
+	}))
+}

proxy/mtproto/config.go

@@ -0,0 +1,24 @@
+package mtproto
+
+import (
+	"v2ray.com/core/common/protocol"
+)
+
+func (a *Account) Equals(another protocol.Account) bool {
+	aa, ok := another.(*Account)
+	if !ok {
+		return false
+	}
+
+	if len(a.Secret) != len(aa.Secret) {
+		return false
+	}
+
+	for i, v := range a.Secret {
+		if v != aa.Secret[i] {
+			return false
+		}
+	}
+
+	return true
+}

proxy/mtproto/config.pb.go

@@ -0,0 +1,153 @@
+package mtproto
+
+import proto "github.com/golang/protobuf/proto"
+import fmt "fmt"
+import math "math"
+import protocol "v2ray.com/core/common/protocol"
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
+
+type Account struct {
+	Secret               []byte   `protobuf:"bytes,1,opt,name=secret,proto3" json:"secret,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Account) Reset()         { *m = Account{} }
+func (m *Account) String() string { return proto.CompactTextString(m) }
+func (*Account) ProtoMessage()    {}
+func (*Account) Descriptor() ([]byte, []int) {
+	return fileDescriptor_config_32dc1a2aa94bc2a8, []int{0}
+}
+func (m *Account) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Account.Unmarshal(m, b)
+}
+func (m *Account) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Account.Marshal(b, m, deterministic)
+}
+func (dst *Account) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Account.Merge(dst, src)
+}
+func (m *Account) XXX_Size() int {
+	return xxx_messageInfo_Account.Size(m)
+}
+func (m *Account) XXX_DiscardUnknown() {
+	xxx_messageInfo_Account.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Account proto.InternalMessageInfo
+
+func (m *Account) GetSecret() []byte {
+	if m != nil {
+		return m.Secret
+	}
+	return nil
+}
+
+type ServerConfig struct {
+	// User is a list of users that allowed to connect to this inbound.
+	// Although this is a repeated field, only the first user is effective for now.
+	User                 []*protocol.User `protobuf:"bytes,1,rep,name=user,proto3" json:"user,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}         `json:"-"`
+	XXX_unrecognized     []byte           `json:"-"`
+	XXX_sizecache        int32            `json:"-"`
+}
+
+func (m *ServerConfig) Reset()         { *m = ServerConfig{} }
+func (m *ServerConfig) String() string { return proto.CompactTextString(m) }
+func (*ServerConfig) ProtoMessage()    {}
+func (*ServerConfig) Descriptor() ([]byte, []int) {
+	return fileDescriptor_config_32dc1a2aa94bc2a8, []int{1}
+}
+func (m *ServerConfig) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_ServerConfig.Unmarshal(m, b)
+}
+func (m *ServerConfig) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_ServerConfig.Marshal(b, m, deterministic)
+}
+func (dst *ServerConfig) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ServerConfig.Merge(dst, src)
+}
+func (m *ServerConfig) XXX_Size() int {
+	return xxx_messageInfo_ServerConfig.Size(m)
+}
+func (m *ServerConfig) XXX_DiscardUnknown() {
+	xxx_messageInfo_ServerConfig.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ServerConfig proto.InternalMessageInfo
+
+func (m *ServerConfig) GetUser() []*protocol.User {
+	if m != nil {
+		return m.User
+	}
+	return nil
+}
+
+type ClientConfig struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ClientConfig) Reset()         { *m = ClientConfig{} }
+func (m *ClientConfig) String() string { return proto.CompactTextString(m) }
+func (*ClientConfig) ProtoMessage()    {}
+func (*ClientConfig) Descriptor() ([]byte, []int) {
+	return fileDescriptor_config_32dc1a2aa94bc2a8, []int{2}
+}
+func (m *ClientConfig) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_ClientConfig.Unmarshal(m, b)
+}
+func (m *ClientConfig) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_ClientConfig.Marshal(b, m, deterministic)
+}
+func (dst *ClientConfig) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ClientConfig.Merge(dst, src)
+}
+func (m *ClientConfig) XXX_Size() int {
+	return xxx_messageInfo_ClientConfig.Size(m)
+}
+func (m *ClientConfig) XXX_DiscardUnknown() {
+	xxx_messageInfo_ClientConfig.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ClientConfig proto.InternalMessageInfo
+
+func init() {
+	proto.RegisterType((*Account)(nil), "v2ray.core.proxy.mtproto.Account")
+	proto.RegisterType((*ServerConfig)(nil), "v2ray.core.proxy.mtproto.ServerConfig")
+	proto.RegisterType((*ClientConfig)(nil), "v2ray.core.proxy.mtproto.ClientConfig")
+}
+
+func init() {
+	proto.RegisterFile("v2ray.com/core/proxy/mtproto/config.proto", fileDescriptor_config_32dc1a2aa94bc2a8)
+}
+
+var fileDescriptor_config_32dc1a2aa94bc2a8 = []byte{
+	// 221 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x8f, 0xc1, 0x4a, 0xc4, 0x30,
+	0x10, 0x86, 0x89, 0xca, 0x2e, 0xc4, 0xe2, 0xa1, 0x07, 0x09, 0xe2, 0xa1, 0xf6, 0xb4, 0x5e, 0x26,
+	0x50, 0x7d, 0x01, 0xed, 0x5e, 0x85, 0xa5, 0xa2, 0x07, 0x6f, 0xeb, 0x30, 0xca, 0xc2, 0x26, 0x53,
+	0xa6, 0x69, 0xb1, 0xaf, 0xe4, 0x53, 0x4a, 0x93, 0x16, 0x44, 0xf0, 0x94, 0xfc, 0xfc, 0x1f, 0xdf,
+	0x9f, 0xe8, 0xdb, 0xa1, 0x92, 0xfd, 0x08, 0xc8, 0xce, 0x22, 0x0b, 0xd9, 0x56, 0xf8, 0x6b, 0xb4,
+	0x2e, 0xb4, 0xc2, 0x81, 0x2d, 0xb2, 0xff, 0x38, 0x7c, 0x42, 0x0c, 0xb9, 0x59, 0x50, 0x21, 0x88,
+	0x18, 0xcc, 0xd8, 0xd5, 0x5f, 0x09, 0xb2, 0x73, 0xec, 0x6d, 0x2c, 0x91, 0x8f, 0xb6, 0xef, 0x48,
+	0x92, 0xa4, 0xbc, 0xd1, 0xeb, 0x07, 0x44, 0xee, 0x7d, 0xc8, 0x2f, 0xf5, 0xaa, 0x23, 0x14, 0x0a,
+	0x46, 0x15, 0x6a, 0x93, 0x35, 0x73, 0x2a, 0xb7, 0x3a, 0x7b, 0x26, 0x19, 0x48, 0xea, 0xb8, 0x9e,
+	0xdf, 0xeb, 0xb3, 0x49, 0x60, 0x54, 0x71, 0xba, 0x39, 0xaf, 0x0a, 0xf8, 0xf5, 0x8c, 0x34, 0x04,
+	0xcb, 0x10, 0xbc, 0x74, 0x24, 0x4d, 0xa4, 0xcb, 0x0b, 0x9d, 0xd5, 0xc7, 0x03, 0xf9, 0x90, 0x2c,
+	0x8f, 0x5b, 0x7d, 0x8d, 0xec, 0xe0, 0xbf, 0x3f, 0xec, 0xd4, 0xdb, 0x7a, 0xbe, 0x7e, 0x9f, 0x98,
+	0xd7, 0xaa, 0xd9, 0x8f, 0x50, 0x4f, 0xd4, 0x2e, 0x52, 0x4f, 0xa9, 0x7a, 0x5f, 0xc5, 0xe3, 0xee,
+	0x27, 0x00, 0x00, 0xff, 0xff, 0x54, 0x23, 0xa0, 0xae, 0x37, 0x01, 0x00, 0x00,
+}

proxy/mtproto/config.proto

@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package v2ray.core.proxy.mtproto;
+option csharp_namespace = "V2Ray.Core.Proxy.Mtproto";
+option go_package = "mtproto";
+option java_package = "com.v2ray.core.proxy.mtproto";
+option java_multiple_files = true;
+
+import "v2ray.com/core/common/protocol/user.proto";
+
+message Account {
+  bytes secret = 1;
+}
+
+message ServerConfig {
+  // User is a list of users that allowed to connect to this inbound.
+  // Although this is a repeated field, only the first user is effective for now.
+  repeated v2ray.core.common.protocol.User user = 1;
+}
+
+message ClientConfig {
+
+}

proxy/mtproto/errors.generated.go

@@ -0,0 +1,5 @@
+package mtproto
+
+import "v2ray.com/core/common/errors"
+
+func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("Proxy", "MTProto") }

proxy/mtproto/mtproto.go

@@ -0,0 +1,3 @@
+package mtproto
+
+//go:generate go run $GOPATH/src/v2ray.com/core/common/errors/errorgen/main.go -pkg mtproto -path Proxy,MTProto

proxy/mtproto/server.go

@@ -0,0 +1,110 @@
+package mtproto
+
+import (
+	"context"
+
+	"v2ray.com/core"
+	"v2ray.com/core/common"
+	"v2ray.com/core/common/buf"
+	"v2ray.com/core/common/crypto"
+	"v2ray.com/core/common/net"
+	"v2ray.com/core/common/protocol"
+	"v2ray.com/core/common/task"
+	"v2ray.com/core/transport/internet"
+	"v2ray.com/core/transport/pipe"
+)
+
+var (
+	dcList = []net.Address{
+		net.ParseAddress("149.154.175.50"),
+		net.ParseAddress("149.154.167.51"),
+		net.ParseAddress("149.154.175.100"),
+		net.ParseAddress("149.154.167.91"),
+		net.ParseAddress("149.154.171.5"),
+	}
+)
+
+type Server struct {
+	user    *protocol.User
+	account *Account
+}
+
+func NewServer(ctx context.Context, config *ServerConfig) (*Server, error) {
+	if len(config.User) == 0 {
+		return nil, newError("no user configured.")
+	}
+
+	user := config.User[0]
+	rawAccount, err := config.User[0].GetTypedAccount()
+	if err != nil {
+		return nil, newError("invalid account").Base(err)
+	}
+	account, ok := rawAccount.(*Account)
+	if !ok {
+		return nil, newError("not a MTProto account")
+	}
+
+	return &Server{
+		user:    user,
+		account: account,
+	}, nil
+}
+
+func (s *Server) Network() net.NetworkList {
+	return net.NetworkList{
+		Network: []net.Network{net.Network_TCP},
+	}
+}
+
+func (s *Server) Process(ctx context.Context, network net.Network, conn internet.Connection, dispatcher core.Dispatcher) error {
+	auth, err := ReadAuthentication(conn)
+	if err != nil {
+		return newError("failed to read authentication header").Base(err)
+	}
+
+	auth.ApplySecret(s.account.Secret)
+
+	decryptor := crypto.NewAesCTRStream(auth.DecodingKey[:], auth.DecodingNonce[:])
+	decryptor.XORKeyStream(auth.Header[:], auth.Header[:])
+
+	dcID := auth.DataCenterID()
+	if dcID >= uint16(len(dcList)) {
+		return newError("invalid data center id: ", dcID)
+	}
+
+	dest := net.Destination{
+		Network: net.Network_TCP,
+		Address: dcList[dcID],
+		Port:    net.Port(443),
+	}
+	link, err := dispatcher.Dispatch(ctx, dest)
+	if err != nil {
+		return newError("failed to dispatch request to: ", dest).Base(err)
+	}
+
+	request := func() error {
+		reader := buf.NewReader(crypto.NewCryptionReader(decryptor, conn))
+		return buf.Copy(reader, link.Writer)
+	}
+
+	response := func() error {
+		encryptor := crypto.NewAesCTRStream(auth.EncodingKey[:], auth.EncodingNonce[:])
+		writer := buf.NewWriter(crypto.NewCryptionWriter(encryptor, conn))
+		return buf.Copy(link.Reader, writer)
+	}
+
+	var responseDoneAndCloseWriter = task.Single(response, task.OnSuccess(task.Close(link.Writer)))
+	if err := task.Run(task.WithContext(ctx), task.Parallel(request, responseDoneAndCloseWriter))(); err != nil {
+		pipe.CloseError(link.Reader)
+		pipe.CloseError(link.Writer)
+		return newError("connection ends").Base(err)
+	}
+
+	return nil
+}
+
+func init() {
+	common.Must(common.RegisterConfig((*ServerConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
+		return NewServer(ctx, config.(*ServerConfig))
+	}))
+}