From d10729490d11b1065be31973fe540713fcd9103d Mon Sep 17 00:00:00 2001 From: 9seconds Date: Thu, 10 Oct 2019 17:49:16 +0300 Subject: [PATCH] Remove obsolete files --- _antireplay/cache.go | 37 ---- _antireplay/hasher.go | 9 - _client/client.go | 15 -- _client/direct.go | 63 ------- _client/middle.go | 42 ----- _config/config.go | 224 ----------------------- _config/global_ips.go | 52 ------ _config/urls.go | 58 ------ _mtproto/connection_options.go | 87 --------- _mtproto/rpc/handshake_request.go | 26 --- _mtproto/rpc/handshake_response.go | 55 ------ _mtproto/rpc/nonce_request.go | 52 ------ _mtproto/rpc/nonce_response.go | 60 ------ _mtproto/rpc/proxy_flags.go | 59 ------ _mtproto/rpc/proxy_request.go | 105 ----------- _mtproto/rpc/rpc.go | 28 --- _obfuscated2/frame.go | 121 ------------ _obfuscated2/frame_test.go | 106 ----------- _obfuscated2/obfuscated2.go | 81 -------- _obfuscated2/obfuscated2_test.go | 97 ---------- _proxy/proxy.go | 178 ------------------ _stats/channels.go | 76 -------- _stats/init.go | 28 --- _stats/prometheus.go | 91 --------- _stats/server.go | 40 ---- _stats/stats.go | 175 ------------------ _stats/statsd.go | 77 -------- _telegram/dialer.go | 52 ------ _telegram/direct.go | 79 -------- _telegram/middle.go | 139 -------------- _telegram/middle_caller.go | 191 ------------------- _telegram/telegram.go | 68 ------- _utils/read_current_data.go | 21 --- _utils/reverse_bytes.go | 15 -- _utils/uint24.go | 15 -- _wrappers/blockcipher.go | 99 ---------- _wrappers/conn.go | 154 ---------------- _wrappers/mtproto_abridged.go | 159 ---------------- _wrappers/mtproto_cipher.go | 98 ---------- _wrappers/mtproto_frame.go | 161 ---------------- _wrappers/mtproto_intermediate.go | 117 ------------ _wrappers/mtproto_intermediate_secure.go | 74 -------- _wrappers/mtproto_proxy.go | 165 ----------------- _wrappers/streamcipher.go | 73 -------- _wrappers/streamcipher_pool.go | 14 -- _wrappers/wrap.go | 111 ----------- 46 files changed, 3847 deletions(-) delete mode 100644 _antireplay/cache.go delete mode 100644 _antireplay/hasher.go delete mode 100644 _client/client.go delete mode 100644 _client/direct.go delete mode 100644 _client/middle.go delete mode 100644 _config/config.go delete mode 100644 _config/global_ips.go delete mode 100644 _config/urls.go delete mode 100644 _mtproto/connection_options.go delete mode 100644 _mtproto/rpc/handshake_request.go delete mode 100644 _mtproto/rpc/handshake_response.go delete mode 100644 _mtproto/rpc/nonce_request.go delete mode 100644 _mtproto/rpc/nonce_response.go delete mode 100644 _mtproto/rpc/proxy_flags.go delete mode 100644 _mtproto/rpc/proxy_request.go delete mode 100644 _mtproto/rpc/rpc.go delete mode 100644 _obfuscated2/frame.go delete mode 100644 _obfuscated2/frame_test.go delete mode 100644 _obfuscated2/obfuscated2.go delete mode 100644 _obfuscated2/obfuscated2_test.go delete mode 100644 _proxy/proxy.go delete mode 100644 _stats/channels.go delete mode 100644 _stats/init.go delete mode 100644 _stats/prometheus.go delete mode 100644 _stats/server.go delete mode 100644 _stats/stats.go delete mode 100644 _stats/statsd.go delete mode 100644 _telegram/dialer.go delete mode 100644 _telegram/direct.go delete mode 100644 _telegram/middle.go delete mode 100644 _telegram/middle_caller.go delete mode 100644 _telegram/telegram.go delete mode 100644 _utils/read_current_data.go delete mode 100644 _utils/reverse_bytes.go delete mode 100644 _utils/uint24.go delete mode 100644 _wrappers/blockcipher.go delete mode 100644 _wrappers/conn.go delete mode 100644 _wrappers/mtproto_abridged.go delete mode 100644 _wrappers/mtproto_cipher.go delete mode 100644 _wrappers/mtproto_frame.go delete mode 100644 _wrappers/mtproto_intermediate.go delete mode 100644 _wrappers/mtproto_intermediate_secure.go delete mode 100644 _wrappers/mtproto_proxy.go delete mode 100644 _wrappers/streamcipher.go delete mode 100644 _wrappers/streamcipher_pool.go delete mode 100644 _wrappers/wrap.go diff --git a/_antireplay/cache.go b/_antireplay/cache.go deleted file mode 100644 index 4923d52..0000000 --- a/_antireplay/cache.go +++ /dev/null @@ -1,37 +0,0 @@ -package antireplay - -import ( - "github.com/allegro/bigcache" - "github.com/juju/errors" - - "github.com/9seconds/mtg/config" -) - -// Cache defines storage for obfuscated2 handshake frames. -type Cache struct { - cache *bigcache.BigCache -} - -func (a Cache) Add(frame []byte) { - a.cache.Set(string(frame), nil) // nolint: errcheck -} - -func (a Cache) Has(frame []byte) bool { - _, err := a.cache.Get(string(frame)) - - return err == nil -} - -func NewCache(config *config.Config) (Cache, error) { - cache, err := bigcache.NewBigCache(bigcache.Config{ - Shards: 1024, - LifeWindow: config.AntiReplayEvictionTime, - Hasher: hasher{}, - HardMaxCacheSize: config.AntiReplayMaxSize, - }) - if err != nil { - return Cache{}, errors.Annotate(err, "Cannot make cache") - } - - return Cache{cache}, nil -} diff --git a/_antireplay/hasher.go b/_antireplay/hasher.go deleted file mode 100644 index acaf9b6..0000000 --- a/_antireplay/hasher.go +++ /dev/null @@ -1,9 +0,0 @@ -package antireplay - -import "github.com/cespare/xxhash" - -type hasher struct{} - -func (h hasher) Sum64(value string) uint64 { - return xxhash.Sum64String(value) -} diff --git a/_client/client.go b/_client/client.go deleted file mode 100644 index b8f1b93..0000000 --- a/_client/client.go +++ /dev/null @@ -1,15 +0,0 @@ -package client - -import ( - "context" - "net" - - "github.com/9seconds/mtg/antireplay" - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/wrappers" -) - -// Init defines common method for initializing client connections. -type Init func(context.Context, context.CancelFunc, net.Conn, string, - antireplay.Cache, *config.Config) (wrappers.Wrap, *mtproto.ConnectionOpts, error) diff --git a/_client/direct.go b/_client/direct.go deleted file mode 100644 index 5ee0a39..0000000 --- a/_client/direct.go +++ /dev/null @@ -1,63 +0,0 @@ -package client - -import ( - "context" - "net" - "time" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/antireplay" - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/obfuscated2" - "github.com/9seconds/mtg/wrappers" -) - -const handshakeTimeout = 10 * time.Second - -// DirectInit initializes client connection for proxy which connects to -// Telegram directly. -func DirectInit(ctx context.Context, cancel context.CancelFunc, socket net.Conn, - connID string, antiReplayCache antireplay.Cache, - conf *config.Config) (wrappers.Wrap, *mtproto.ConnectionOpts, error) { - tcpSocket := socket.(*net.TCPConn) - if err := tcpSocket.SetNoDelay(false); err != nil { - return nil, nil, errors.Annotate(err, "Cannot disable NO_DELAY to client socket") - } - if err := tcpSocket.SetReadBuffer(conf.ReadBufferSize); err != nil { - return nil, nil, errors.Annotate(err, "Cannot set read buffer size of client socket") - } - if err := tcpSocket.SetWriteBuffer(conf.WriteBufferSize); err != nil { - return nil, nil, errors.Annotate(err, "Cannot set write buffer size of client socket") - } - - socket.SetReadDeadline(time.Now().Add(handshakeTimeout)) // nolint: errcheck, gosec - frame, err := obfuscated2.ExtractFrame(socket) - if err != nil { - return nil, nil, errors.Annotate(err, "Cannot extract frame") - } - socket.SetReadDeadline(time.Time{}) // nolint: errcheck, gosec - - conn := wrappers.NewConn(ctx, cancel, socket, connID, wrappers.ConnPurposeClient, conf.PublicIPv4, conf.PublicIPv6) - obfs2, connOpts, err := obfuscated2.ParseObfuscated2ClientFrame(conf.Secret, frame) - if err != nil { - return nil, nil, errors.Annotate(err, "Cannot parse obfuscated frame") - } - - var replayPart = []byte(frame) - - if antiReplayCache.Has(replayPart[4:60]) { - return nil, nil, errors.New("Replay attack is detected") - } - antiReplayCache.Add(replayPart[4:60]) - - connOpts.ConnectionProto = mtproto.ConnectionProtocolAny - connOpts.ClientAddr = conn.RemoteAddr() - - conn = wrappers.NewStreamCipher(conn, obfs2.Encryptor, obfs2.Decryptor) - - conn.Logger().Infow("Client connection initialized") - - return conn, connOpts, nil -} diff --git a/_client/middle.go b/_client/middle.go deleted file mode 100644 index ecacd8e..0000000 --- a/_client/middle.go +++ /dev/null @@ -1,42 +0,0 @@ -package client - -import ( - "context" - "net" - - "github.com/9seconds/mtg/antireplay" - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/wrappers" -) - -// MiddleInit initializes client connection for proxy which has to -// support promoted channels, connect to Telegram middle proxies etc. -func MiddleInit(ctx context.Context, cancel context.CancelFunc, socket net.Conn, - connID string, antiReplayCache antireplay.Cache, - conf *config.Config) (wrappers.Wrap, *mtproto.ConnectionOpts, error) { - conn, opts, err := DirectInit(ctx, cancel, socket, connID, antiReplayCache, conf) - if err != nil { - return nil, nil, err - } - connStream := conn.(wrappers.StreamReadWriteCloser) - - var newConn wrappers.PacketReadWriteCloser - switch opts.ConnectionType { - case mtproto.ConnectionTypeAbridged: - newConn = wrappers.NewMTProtoAbridged(connStream, opts) - case mtproto.ConnectionTypeIntermediate: - newConn = wrappers.NewMTProtoIntermediate(connStream, opts) - case mtproto.ConnectionTypeSecure: - newConn = wrappers.NewMTProtoIntermediateSecure(connStream, opts) - default: - panic("Unknown connection type") - } - - opts.ConnectionProto = mtproto.ConnectionProtocolIPv4 - if socket.LocalAddr().(*net.TCPAddr).IP.To4() == nil { - opts.ConnectionProto = mtproto.ConnectionProtocolIPv6 - } - - return newConn, opts, err -} diff --git a/_config/config.go b/_config/config.go deleted file mode 100644 index dc43864..0000000 --- a/_config/config.go +++ /dev/null @@ -1,224 +0,0 @@ -package config - -import ( - "bytes" - "encoding/hex" - "fmt" - "net" - "strconv" - "time" - - "github.com/juju/errors" - statsd "gopkg.in/alexcesaro/statsd.v2" -) - -// Config represents common configuration of mtg. -type Config struct { - Debug bool - Verbose bool - SecureMode bool - SecureOnly bool - - ReadBufferSize int - WriteBufferSize int - - BindPort uint16 - PublicIPv4Port uint16 - PublicIPv6Port uint16 - StatsPort uint16 - - BindIP net.IP - PublicIPv4 net.IP - PublicIPv6 net.IP - StatsIP net.IP - - AntiReplayMaxSize int - AntiReplayEvictionTime time.Duration - - StatsD struct { - Addr net.Addr - Prefix string - Tags map[string]string - TagsFormat statsd.TagFormat - Enabled bool - } - Prometheus struct { - Prefix string - } - - Secret []byte - AdTag []byte -} - -// URLs contains links to the proxy (tg://, t.me) and their QR codes. -type URLs struct { - TG string `json:"tg_url"` - TMe string `json:"tme_url"` - TGQRCode string `json:"tg_qrcode"` - TMeQRCode string `json:"tme_qrcode"` -} - -// IPURLs contains links to both ipv4 and ipv6 of the proxy. -type IPURLs struct { - IPv4 URLs `json:"ipv4"` - IPv6 URLs `json:"ipv6"` - BotSecret string `json:"secret_for_mtproxybot"` -} - -// BindAddr returns connection for this server to bind to. -func (c *Config) BindAddr() string { - return getAddr(c.BindIP, c.BindPort) -} - -// StatAddr returns connection string to the stats API. -func (c *Config) StatAddr() string { - return getAddr(c.StatsIP, c.StatsPort) -} - -// UseMiddleProxy defines if this proxy has to connect middle proxies -// which supports promoted channels or directly access Telegram. -func (c *Config) UseMiddleProxy() bool { - return len(c.AdTag) > 0 -} - -// BotSecretString returns secret string which should work with MTProxybot. -func (c *Config) BotSecretString() string { - return hex.EncodeToString(c.Secret) -} - -// SecretString returns a secret in a form entered on the start of the -// application. -func (c *Config) SecretString() string { - secret := c.BotSecretString() - if c.SecureMode { - return "dd" + secret - } - return secret -} - -// GetURLs returns configured IPURLs instance with links to this server. -func (c *Config) GetURLs() IPURLs { - urls := IPURLs{} - secret := c.SecretString() - if c.PublicIPv4 != nil { - urls.IPv4 = getURLs(c.PublicIPv4, c.PublicIPv4Port, secret) - } - if c.PublicIPv6 != nil { - urls.IPv6 = getURLs(c.PublicIPv6, c.PublicIPv6Port, secret) - } - urls.BotSecret = c.BotSecretString() - - return urls -} - -func getAddr(host fmt.Stringer, port uint16) string { - return net.JoinHostPort(host.String(), strconv.Itoa(int(port))) -} - -// NewConfig returns new configuration. If required, it manages and -// fetches data from external sources. Parameters passed to this -// function, should come from command line arguments. -func NewConfig(debug, verbose bool, // nolint: gocyclo - writeBufferSize, readBufferSize uint32, - bindIP, publicIPv4, publicIPv6, statsIP net.IP, - bindPort, publicIPv4Port, publicIPv6Port, statsPort, statsdPort uint16, - statsdIP, statsdNetwork, statsdPrefix, statsdTagsFormat string, - statsdTags map[string]string, prometheusPrefix string, - secureOnly bool, - antiReplayMaxSize int, antiReplayEvictionTime time.Duration, - secret, adtag []byte) (*Config, error) { - secureMode := secureOnly - if bytes.HasPrefix(secret, []byte{0xdd}) && len(secret) == 17 { - secureMode = true - secret = bytes.TrimPrefix(secret, []byte{0xdd}) - } else if len(secret) != 16 { - return nil, errors.New("Telegram demands secret of length 32") - } - - var err error - if publicIPv4 == nil { - publicIPv4, err = getGlobalIPv4() - if err != nil { - publicIPv4 = nil - } else if publicIPv4.To4() == nil { - return nil, errors.Errorf("IP %s is not IPv4", publicIPv4.String()) - } - } - if publicIPv4Port == 0 { - publicIPv4Port = bindPort - } - - if publicIPv6 == nil { - publicIPv6, err = getGlobalIPv6() - if err != nil { - publicIPv6 = nil - } else if publicIPv6.To4() != nil { - return nil, errors.Errorf("IP %s is not IPv6", publicIPv6.String()) - } - } - if publicIPv6Port == 0 { - publicIPv6Port = bindPort - } - - if statsIP == nil { - statsIP = publicIPv4 - } - - conf := &Config{ - Debug: debug, - Verbose: verbose, - SecureOnly: secureOnly, - BindIP: bindIP, - BindPort: bindPort, - PublicIPv4: publicIPv4, - PublicIPv4Port: publicIPv4Port, - PublicIPv6: publicIPv6, - PublicIPv6Port: publicIPv6Port, - StatsIP: statsIP, - StatsPort: statsPort, - Secret: secret, - AdTag: adtag, - SecureMode: secureMode, - ReadBufferSize: int(readBufferSize), - WriteBufferSize: int(writeBufferSize), - AntiReplayMaxSize: antiReplayMaxSize, - AntiReplayEvictionTime: antiReplayEvictionTime, - } - conf.Prometheus.Prefix = prometheusPrefix - - if statsdIP != "" { - conf.StatsD.Enabled = true - conf.StatsD.Prefix = statsdPrefix - conf.StatsD.Tags = statsdTags - - var ( - addr net.Addr - err error - ) - hostPort := net.JoinHostPort(statsdIP, strconv.Itoa(int(statsdPort))) - switch statsdNetwork { - case "tcp": - addr, err = net.ResolveTCPAddr("tcp", hostPort) - case "udp": - addr, err = net.ResolveUDPAddr("udp", hostPort) - default: - err = errors.Errorf("Unknown network %s", statsdNetwork) - } - if err != nil { - return nil, errors.Annotate(err, "Cannot resolve statsd address") - } - conf.StatsD.Addr = addr - - switch statsdTagsFormat { - case "datadog": - conf.StatsD.TagsFormat = statsd.Datadog - case "influxdb": - conf.StatsD.TagsFormat = statsd.InfluxDB - case "": - default: - return nil, errors.Errorf("Unknown tags format %s", statsdTagsFormat) - } - } - - return conf, nil -} diff --git a/_config/global_ips.go b/_config/global_ips.go deleted file mode 100644 index 9a144a6..0000000 --- a/_config/global_ips.go +++ /dev/null @@ -1,52 +0,0 @@ -package config - -import ( - "context" - "io/ioutil" - "net" - "net/http" - "strings" - - "github.com/juju/errors" -) - -const ifconfigAddress = "https://ifconfig.co/ip" - -func getGlobalIPv4() (net.IP, error) { - return fetchIP("tcp4") -} - -func getGlobalIPv6() (net.IP, error) { - return fetchIP("tcp6") -} - -func fetchIP(network string) (net.IP, error) { - dialer := &net.Dialer{FallbackDelay: -1} - client := &http.Client{ - Jar: nil, - Transport: &http.Transport{ - DialContext: func(ctx context.Context, _, addr string) (net.Conn, error) { - return dialer.DialContext(ctx, network, addr) - }, - }, - } - - resp, err := client.Get(ifconfigAddress) - if err != nil { - return nil, err - } - defer resp.Body.Close() // nolint: errcheck - - respDataBytes, err := ioutil.ReadAll(resp.Body) - if err != nil { - return nil, err - } - respData := strings.TrimSpace(string(respDataBytes)) - - ip := net.ParseIP(respData) - if ip == nil { - return nil, errors.Errorf("ifconfig.co returns incorrect IP %s", respData) - } - - return ip, nil -} diff --git a/_config/urls.go b/_config/urls.go deleted file mode 100644 index 5f4e6d8..0000000 --- a/_config/urls.go +++ /dev/null @@ -1,58 +0,0 @@ -package config - -import ( - "net" - "net/url" - "strconv" -) - -func getURLs(addr net.IP, port uint16, secret string) (urls URLs) { - values := url.Values{} - values.Set("server", addr.String()) - values.Set("port", strconv.Itoa(int(port))) - values.Set("secret", secret) - - urls.TG = makeTGURL(values) - urls.TMe = makeTMeURL(values) - urls.TGQRCode = makeQRCodeURL(urls.TG) - urls.TMeQRCode = makeQRCodeURL(urls.TG) - - return -} - -func makeTGURL(values url.Values) string { - tgURL := url.URL{ - Scheme: "tg", - Host: "proxy", - RawQuery: values.Encode(), - } - - return tgURL.String() -} - -func makeTMeURL(values url.Values) string { - tMeURL := url.URL{ - Scheme: "https", - Host: "t.me", - Path: "proxy", - RawQuery: values.Encode(), - } - - return tMeURL.String() -} - -func makeQRCodeURL(data string) string { - qr := url.URL{ - Scheme: "https", - Host: "api.qrserver.com", - Path: "v1/create-qr-code", - } - - values := url.Values{} - values.Set("qzone", "4") - values.Set("format", "svg") - values.Set("data", data) - qr.RawQuery = values.Encode() - - return qr.String() -} diff --git a/_mtproto/connection_options.go b/_mtproto/connection_options.go deleted file mode 100644 index 0230f5f..0000000 --- a/_mtproto/connection_options.go +++ /dev/null @@ -1,87 +0,0 @@ -package mtproto - -import ( - "bytes" - "net" - - "github.com/juju/errors" -) - -// ConnectionType is a type of obfuscated2/mtproto connection requested -// by the user. -type ConnectionType uint8 - -// ConnectionProtocol is a type of IP protocol to use. -type ConnectionProtocol uint8 - -// Hacks is a simple structure to store flags for packet transmission. -type Hacks struct { - SimpleAck bool - QuickAck bool -} - -// ConnectionOpts presents an options, metadata on connection requested -// by the user on handshake. -type ConnectionOpts struct { - DC int16 - ConnectionType ConnectionType - ConnectionProto ConnectionProtocol - // Read and Write means direction related to the client. - // ReadHacks are meant to be flushed on client read - // WriteHacks are meant to be flushed on client write. - ReadHacks Hacks - WriteHacks Hacks - ClientAddr *net.TCPAddr -} - -// Different connection types which user requests from Telegram. -const ( - ConnectionTypeUnknown ConnectionType = iota - ConnectionTypeAbridged - ConnectionTypeIntermediate - ConnectionTypeSecure -) - -// ConnectionProtocol* define which connection protocols to use. -// ConnectionProtocolAny means that any is suitable. -const ( - ConnectionProtocolIPv4 ConnectionProtocol = 1 - ConnectionProtocolIPv6 = ConnectionProtocolIPv4 << 1 - ConnectionProtocolAny = ConnectionProtocolIPv4 | ConnectionProtocolIPv6 -) - -// Connection tags for mtproto handshakes. -var ( - ConnectionTagAbridged = []byte{0xef, 0xef, 0xef, 0xef} - ConnectionTagIntermediate = []byte{0xee, 0xee, 0xee, 0xee} - ConnectionTagSecure = []byte{0xdd, 0xdd, 0xdd, 0xdd} -) - -// Tag maps connection type to the corresponding handshake tag. -func (t ConnectionType) Tag() ([]byte, error) { - switch t { - case ConnectionTypeAbridged: - return ConnectionTagAbridged, nil - case ConnectionTypeIntermediate: - return ConnectionTagIntermediate, nil - case ConnectionTypeSecure: - return ConnectionTagSecure, nil - default: - return nil, errors.Errorf("Unknown connection type %d", t) - } -} - -// ConnectionTagFromHandshake maps magic bytes to the connection type. -func ConnectionTagFromHandshake(magic []byte) (ConnectionType, error) { - if bytes.Equal(magic, ConnectionTagIntermediate) { - return ConnectionTypeIntermediate, nil - } - if bytes.Equal(magic, ConnectionTagAbridged) { - return ConnectionTypeAbridged, nil - } - if bytes.Equal(magic, ConnectionTagSecure) { - return ConnectionTypeSecure, nil - } - - return ConnectionTypeUnknown, errors.New("Unknown handshake protocol") -} diff --git a/_mtproto/rpc/handshake_request.go b/_mtproto/rpc/handshake_request.go deleted file mode 100644 index dcd2290..0000000 --- a/_mtproto/rpc/handshake_request.go +++ /dev/null @@ -1,26 +0,0 @@ -package rpc - -import "bytes" - -// HandshakeRequest is the data type which is responsible for -// constructing of correct handshake request. -type HandshakeRequest struct { -} - -// Bytes returns serialized handshake request. -func (r *HandshakeRequest) Bytes() []byte { - buf := &bytes.Buffer{} - buf.Grow(len(TagHandshake) + len(HandshakeFlags) + len(HandshakeSenderPID) + len(HandshakePeerPID)) - - buf.Write(TagHandshake) // nolint: gosec - buf.Write(HandshakeFlags) // nolint: gosec - buf.Write(HandshakeSenderPID) // nolint: gosec - buf.Write(HandshakePeerPID) // nolint: gosec - - return buf.Bytes() -} - -// NewHandshakeRequest creates new HandshakeRequest instance. -func NewHandshakeRequest() *HandshakeRequest { - return &HandshakeRequest{} -} diff --git a/_mtproto/rpc/handshake_response.go b/_mtproto/rpc/handshake_response.go deleted file mode 100644 index 962e949..0000000 --- a/_mtproto/rpc/handshake_response.go +++ /dev/null @@ -1,55 +0,0 @@ -package rpc - -import ( - "bytes" - - "github.com/juju/errors" -) - -// HandshakeResponse defines data structure which is used for storage of -// handshake response. -type HandshakeResponse struct { - Type []byte - Flags []byte - SenderPID []byte - PeerPID []byte -} - -// Bytes returns a serialized handshake response. -func (r *HandshakeResponse) Bytes() []byte { - buf := &bytes.Buffer{} - - buf.Write(r.Type) // nolint: gosec - buf.Write(r.Flags) // nolint: gosec - buf.Write(r.SenderPID) // nolint: gosec - buf.Write(r.PeerPID) // nolint: gosec - - return buf.Bytes() -} - -// Valid checks that handshake response compliments request. -func (r *HandshakeResponse) Valid(req *HandshakeRequest) error { - if !bytes.Equal(r.Type, TagHandshake) { - return errors.New("Unexpected handshake tag") - } - if !bytes.Equal(r.PeerPID, HandshakeSenderPID) { - return errors.New("Incorrect sender PID") - } - - return nil -} - -// NewHandshakeResponse constructs new handshake response from the given -// data. -func NewHandshakeResponse(data []byte) (*HandshakeResponse, error) { - if len(data) != 32 { - return nil, errors.New("Incorrect handshake response length") - } - - return &HandshakeResponse{ - Type: data[:4], - Flags: data[4:8], - SenderPID: data[8:20], - PeerPID: data[20:], - }, nil -} diff --git a/_mtproto/rpc/nonce_request.go b/_mtproto/rpc/nonce_request.go deleted file mode 100644 index 60cf218..0000000 --- a/_mtproto/rpc/nonce_request.go +++ /dev/null @@ -1,52 +0,0 @@ -package rpc - -import ( - "bytes" - "crypto/rand" - "encoding/binary" - "time" - - "github.com/juju/errors" -) - -// NonceRequest is the data type which contains all the data for correct -// nonce request. -type NonceRequest struct { - KeySelector []byte - CryptoTS []byte - Nonce []byte -} - -// Bytes returns serialized nonce request. -func (r *NonceRequest) Bytes() []byte { - buf := &bytes.Buffer{} - - buf.Write(TagNonce) // nolint: gosec - buf.Write(r.KeySelector) // nolint: gosec - buf.Write(NonceCryptoAES) // nolint: gosec - buf.Write(r.CryptoTS) // nolint: gosec - buf.Write(r.Nonce) // nolint: gosec - - return buf.Bytes() -} - -// NewNonceRequest builds new none request based on proxy secret. -func NewNonceRequest(proxySecret []byte) (*NonceRequest, error) { - nonce := make([]byte, 16) - keySelector := make([]byte, 4) - cryptoTS := make([]byte, 4) - - if _, err := rand.Read(nonce); err != nil { - return nil, errors.Annotate(err, "Cannot generate nonce") - } - copy(keySelector, proxySecret) - - timestamp := time.Now().Truncate(time.Second).Unix() % 4294967296 // 256 ^ 4 - do not know how to name - binary.LittleEndian.PutUint32(cryptoTS, uint32(timestamp)) - - return &NonceRequest{ - KeySelector: keySelector, - CryptoTS: cryptoTS, - Nonce: nonce, - }, nil -} diff --git a/_mtproto/rpc/nonce_response.go b/_mtproto/rpc/nonce_response.go deleted file mode 100644 index f7c592e..0000000 --- a/_mtproto/rpc/nonce_response.go +++ /dev/null @@ -1,60 +0,0 @@ -package rpc - -import ( - "bytes" - - "github.com/juju/errors" -) - -// NonceResponse is the data type which contains data of nonce response. -type NonceResponse struct { - NonceRequest - - Type []byte - Crypto []byte -} - -// Bytes returns serialized form of the nonce response. -func (r *NonceResponse) Bytes() []byte { - buf := &bytes.Buffer{} - - buf.Write(r.Type) // nolint: gosec - buf.Write(r.KeySelector) // nolint: gosec - buf.Write(r.Crypto) // nolint: gosec - buf.Write(r.CryptoTS) // nolint: gosec - buf.Write(r.Nonce) // nolint: gosec - - return buf.Bytes() -} - -// Valid checks that nonce response compliments nonce request. -func (r *NonceResponse) Valid(req *NonceRequest) error { - if !bytes.Equal(r.Type, TagNonce) { - return errors.New("Unexpected RPC type") - } - if !bytes.Equal(r.Crypto, NonceCryptoAES) { - return errors.New("Unexpected crypto type") - } - if !bytes.Equal(r.KeySelector, req.KeySelector) { - return errors.New("Unexpected key selector") - } - - return nil -} - -// NewNonceResponse build new nonce response based on the given data. -func NewNonceResponse(data []byte) (*NonceResponse, error) { - if len(data) != 32 { - return nil, errors.New("Unexpected message length") - } - - return &NonceResponse{ - NonceRequest: NonceRequest{ - KeySelector: data[4:8], - CryptoTS: data[12:16], - Nonce: data[16:], - }, - Type: data[:4], - Crypto: data[8:12], - }, nil -} diff --git a/_mtproto/rpc/proxy_flags.go b/_mtproto/rpc/proxy_flags.go deleted file mode 100644 index c59d33f..0000000 --- a/_mtproto/rpc/proxy_flags.go +++ /dev/null @@ -1,59 +0,0 @@ -package rpc - -import ( - "encoding/binary" - "strings" -) - -type proxyRequestFlags uint32 - -const ( - proxyRequestFlagsHasAdTag proxyRequestFlags = 0x8 - proxyRequestFlagsEncrypted proxyRequestFlags = 0x2 - proxyRequestFlagsMagic proxyRequestFlags = 0x1000 - proxyRequestFlagsExtMode2 proxyRequestFlags = 0x20000 - proxyRequestFlagsIntermediate proxyRequestFlags = 0x20000000 - proxyRequestFlagsAbdridged proxyRequestFlags = 0x40000000 - proxyRequestFlagsQuickAck proxyRequestFlags = 0x80000000 - proxyRequestFlagsPad proxyRequestFlags = 0x8000000 -) - -var proxyRequestFlagsEncryptedPrefix [8]byte - -func (r proxyRequestFlags) Bytes() []byte { - converted := make([]byte, 4) - binary.LittleEndian.PutUint32(converted, uint32(r)) - - return converted -} - -func (r proxyRequestFlags) String() string { - flags := make([]string, 0, 7) - - if r&proxyRequestFlagsHasAdTag != 0 { - flags = append(flags, "HAS_AD_TAG") - } - if r&proxyRequestFlagsEncrypted != 0 { - flags = append(flags, "ENCRYPTED") - } - if r&proxyRequestFlagsMagic != 0 { - flags = append(flags, "MAGIC") - } - if r&proxyRequestFlagsExtMode2 != 0 { - flags = append(flags, "EXT_MODE_2") - } - if r&proxyRequestFlagsIntermediate != 0 { - flags = append(flags, "INTERMEDIATE") - } - if r&proxyRequestFlagsAbdridged != 0 { - flags = append(flags, "ABRIDGED") - } - if r&proxyRequestFlagsQuickAck != 0 { - flags = append(flags, "QUICK_ACK") - } - if r&proxyRequestFlagsPad != 0 { - flags = append(flags, "PAD") - } - - return strings.Join(flags, " | ") -} diff --git a/_mtproto/rpc/proxy_request.go b/_mtproto/rpc/proxy_request.go deleted file mode 100644 index bb9e68f..0000000 --- a/_mtproto/rpc/proxy_request.go +++ /dev/null @@ -1,105 +0,0 @@ -package rpc - -import ( - "bytes" - "crypto/rand" - "encoding/binary" - "fmt" - "net" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/mtproto" -) - -// ProxyRequest is the data type for storing data required to compose -// RPC_PROXY_REQ request. -type ProxyRequest struct { - Flags proxyRequestFlags - ConnectionID []byte - OurIPPort []byte - ClientIPPort []byte - ADTag []byte - Options *mtproto.ConnectionOpts -} - -// MakeHeader makes RPC_PROXY_REQ header. We need only to append the -// data for it. -func (r *ProxyRequest) MakeHeader(message []byte) (*bytes.Buffer, fmt.Stringer) { - bufferLength := len(TagProxyRequest) + - 4 + // len(flags) - len(r.ConnectionID) + - len(r.ClientIPPort) + - len(r.OurIPPort) + - len(ProxyRequestExtraSize) + - len(ProxyRequestProxyTag) + - 1 + // len(AdTag) - len(r.ADTag) - bufferLength += bufferLength % 4 - - buf := &bytes.Buffer{} - buf.Grow(bufferLength + len(message)) - - flags := r.Flags - if r.Options.ReadHacks.QuickAck { - flags |= proxyRequestFlagsQuickAck - } - - if bytes.HasPrefix(message, proxyRequestFlagsEncryptedPrefix[:]) { - flags |= proxyRequestFlagsEncrypted - } - - buf.Write(TagProxyRequest) // nolint: gosec - buf.Write(flags.Bytes()) // nolint: gosec - buf.Write(r.ConnectionID) // nolint: gosec - buf.Write(r.ClientIPPort) // nolint: gosec - buf.Write(r.OurIPPort) // nolint: gosec - buf.Write(ProxyRequestExtraSize) // nolint: gosec - buf.Write(ProxyRequestProxyTag) // nolint: gosec - buf.WriteByte(byte(len(r.ADTag))) // nolint: gosec - buf.Write(r.ADTag) // nolint: gosec - buf.Write(make([]byte, (4-buf.Len()%4)%4)) // nolint: gosec - - return buf, flags -} - -// NewProxyRequest build new ProxyRequest data structure. -func NewProxyRequest(clientAddr, ownAddr *net.TCPAddr, - opts *mtproto.ConnectionOpts, adTag []byte) (*ProxyRequest, error) { - flags := proxyRequestFlagsHasAdTag | proxyRequestFlagsMagic | proxyRequestFlagsExtMode2 - - switch opts.ConnectionType { - case mtproto.ConnectionTypeAbridged: - flags |= proxyRequestFlagsAbdridged - case mtproto.ConnectionTypeIntermediate: - flags |= proxyRequestFlagsIntermediate - case mtproto.ConnectionTypeSecure: - flags |= proxyRequestFlagsIntermediate | proxyRequestFlagsPad - default: - panic("Unknown connection type") - } - - request := &ProxyRequest{ - Flags: flags, - ADTag: adTag, - Options: opts, - ConnectionID: make([]byte, 8), - ClientIPPort: make([]byte, 16+4), - OurIPPort: make([]byte, 16+4), - } - - if _, err := rand.Read(request.ConnectionID); err != nil { - return nil, errors.Annotate(err, "Cannot generate connection ID") - } - - port := [4]byte{} - copy(request.ClientIPPort[:16], clientAddr.IP.To16()) - binary.LittleEndian.PutUint32(port[:], uint32(clientAddr.Port)) - copy(request.ClientIPPort[16:], port[:]) - - copy(request.OurIPPort[:16], ownAddr.IP.To16()) - binary.LittleEndian.PutUint32(port[:], uint32(ownAddr.Port)) - copy(request.OurIPPort[16:], port[:]) - - return request, nil -} diff --git a/_mtproto/rpc/rpc.go b/_mtproto/rpc/rpc.go deleted file mode 100644 index 59512a3..0000000 --- a/_mtproto/rpc/rpc.go +++ /dev/null @@ -1,28 +0,0 @@ -package rpc - -// SeqNo* is the number of the sequence which have special meaning for -// the Telegram. -const ( - SeqNoNonce = -2 - SeqNoHandshake = -1 -) - -// Different constants for RPC protocol -var ( - TagCloseExt = []byte{0xa2, 0x34, 0xb6, 0x5e} - TagProxyAns = []byte{0x0d, 0xda, 0x03, 0x44} - TagSimpleAck = []byte{0x9b, 0x40, 0xac, 0x3b} - TagHandshake = []byte{0xf5, 0xee, 0x82, 0x76} - TagNonce = []byte{0xaa, 0x87, 0xcb, 0x7a} - TagProxyRequest = []byte{0xee, 0xf1, 0xce, 0x36} - - NonceCryptoAES = []byte{0x01, 0x00, 0x00, 0x00} - - HandshakeFlags = []byte{0x00, 0x00, 0x00, 0x00} - - ProxyRequestExtraSize = []byte{0x18, 0x00, 0x00, 0x00} - ProxyRequestProxyTag = []byte{0xae, 0x26, 0x1e, 0xdb} - - HandshakeSenderPID = []byte("IPIPPRPDTIME") - HandshakePeerPID = []byte("IPIPPRPDTIME") -) diff --git a/_obfuscated2/frame.go b/_obfuscated2/frame.go deleted file mode 100644 index 33a8dd4..0000000 --- a/_obfuscated2/frame.go +++ /dev/null @@ -1,121 +0,0 @@ -package obfuscated2 - -import ( - "bytes" - "crypto/rand" - "encoding/binary" - "io" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/mtproto" -) - -// [frameOffsetFirst:frameOffsetKey:frameOffsetIV:frameOffsetMagic:frameOffsetDC:frameOffsetEnd] -const ( - frameLenKey = 32 - frameLenIV = 16 - frameLenMagic = 4 - frameLenDC = 2 - - frameOffsetFirst = 8 - frameOffsetKey = frameOffsetFirst + frameLenKey - frameOffsetIV = frameOffsetKey + frameLenIV - frameOffsetMagic = frameOffsetIV + frameLenMagic - frameOffsetDC = frameOffsetMagic + frameLenDC - - FrameLen = 64 -) - -// Frame represents handshake frame. Telegram sends 64 bytes of obfuscated2 -// initialization data first. -// https://blog.susanka.eu/how-telegram-obfuscates-its-mtproto-traffic/ -type Frame []byte - -// Key returns AES encryption key. -func (f Frame) Key() []byte { - return f[frameOffsetFirst:frameOffsetKey] -} - -// IV returns AES encryption initialization vector -func (f Frame) IV() []byte { - return f[frameOffsetKey:frameOffsetIV] -} - -// Magic returns magic bytes from last 8 bytes of frame. Telegram checks -// for values there. If after decryption magic is not as expected, -// connection considered as failed. -func (f Frame) Magic() []byte { - return f[frameOffsetIV:frameOffsetMagic] -} - -// DC returns number of datacenter IP client wants to use. -func (f Frame) DC() (n int16) { - buf := bytes.NewReader(f[frameOffsetMagic:frameOffsetDC]) - if err := binary.Read(buf, binary.LittleEndian, &n); err != nil { - n = 1 - } - - return -} - -// ConnectionType identifies connection type of the handshake frame. -func (f Frame) ConnectionType() (mtproto.ConnectionType, error) { - return mtproto.ConnectionTagFromHandshake(f.Magic()) -} - -// Invert inverts frame for extracting encryption keys. Pkease check that link: -// https://blog.susanka.eu/how-telegram-obfuscates-its-mtproto-traffic/ -func (f Frame) Invert() Frame { - reversed := make(Frame, FrameLen) - copy(reversed, f) - - for i := 0; i < frameLenKey+frameLenIV; i++ { - reversed[frameOffsetFirst+i] = f[frameOffsetIV-1-i] - } - - return reversed -} - -// ExtractFrame extracts exact obfuscated2 handshake frame from given reader. -func ExtractFrame(conn io.Reader) (Frame, error) { - frame := make(Frame, FrameLen) - buf := bytes.NewBuffer(frame) - buf.Reset() - - if _, err := io.CopyN(buf, conn, FrameLen); err != nil { - return nil, errors.Annotate(err, "Cannot extract obfuscated header") - } - copy(frame, buf.Bytes()) - - return frame, nil -} - -func generateFrame(connectionType mtproto.ConnectionType) Frame { - frame := make(Frame, FrameLen) - - for { - if _, err := rand.Read(frame); err != nil { - continue - } - if frame[0] == 0xef { - continue - } - - val := (uint32(frame[3]) << 24) | (uint32(frame[2]) << 16) | (uint32(frame[1]) << 8) | uint32(frame[0]) - if val == 0x44414548 || val == 0x54534f50 || val == 0x20544547 || val == 0x4954504f || val == 0xeeeeeeee { - continue - } - - val = (uint32(frame[7]) << 24) | (uint32(frame[6]) << 16) | (uint32(frame[5]) << 8) | uint32(frame[4]) - if val == 0x00000000 { - continue - } - - // error has to be checked before calling this function - tag, _ := connectionType.Tag() // nolint: errcheck, gosec - copy(frame.Magic(), tag) - - return frame - } -} diff --git a/_obfuscated2/frame_test.go b/_obfuscated2/frame_test.go deleted file mode 100644 index 0983958..0000000 --- a/_obfuscated2/frame_test.go +++ /dev/null @@ -1,106 +0,0 @@ -package obfuscated2 - -import ( - "bytes" - "strconv" - "testing" - - "github.com/stretchr/testify/assert" - - "github.com/9seconds/mtg/mtproto" -) - -func TestFrameKey(t *testing.T) { - toCompare := make([]byte, 32) - for i := 0; i < 32; i++ { - toCompare[i] = byte(1) - } - - assert.Equal(t, toCompare, makeFrame().Key()) -} - -func TestFrameIV(t *testing.T) { - toCompare := make([]byte, 16) - for i := 0; i < 16; i++ { - toCompare[i] = byte(2) - } - - assert.Equal(t, toCompare, makeFrame().IV()) -} - -func TestFrameMagic(t *testing.T) { - toCompare := make([]byte, 4) - for i := 0; i < 4; i++ { - toCompare[i] = 0xee - } - - assert.Equal(t, toCompare, makeFrame().Magic()) -} - -func TestFrameDC(t *testing.T) { - assert.Equal(t, int16(771), makeFrame().DC()) -} - -func TestFrameValid(t *testing.T) { - frame := makeFrame() - connType, err := frame.ConnectionType() - assert.Nil(t, err) - assert.Equal(t, connType, mtproto.ConnectionTypeIntermediate) - - frame[8+32+16+2] = byte(3) - _, err = frame.ConnectionType() - assert.NotNil(t, err) -} - -func TestFrameDoubleInvert(t *testing.T) { - frame := makeFrame() - assert.True(t, bytes.Equal(frame, frame.Invert().Invert())) -} - -func TestFrameInvert(t *testing.T) { - frame := makeFrame() - reversed := frame.Invert() - - assert.Exactly(t, frame[:8], reversed[:8]) - assert.Exactly(t, frame[56:], reversed[56:]) - - toCompare := make([]byte, 48) - for i := 0; i < 48; i++ { - toCompare[i] = frame[55-i] - } - assert.Equal(t, []byte(reversed[8:56]), toCompare) -} - -func TestFrameGenerateValid(t *testing.T) { - validTests := []mtproto.ConnectionType{ - mtproto.ConnectionTypeIntermediate, - mtproto.ConnectionTypeAbridged, - } - for _, test := range validTests { - t.Run(strconv.Itoa(int(test)), func(tt *testing.T) { - frame := generateFrame(test) // nolint: scopelint - conType, err := frame.ConnectionType() - assert.Nil(tt, err) - assert.Equal(tt, conType, test) // nolint: scopelint - }) - } -} - -func makeFrame() Frame { - f := make(Frame, FrameLen) - - for i := 8; i < (8 + 32); i++ { - f[i] = byte(1) - } - for i := (8 + 32); i < (8 + 32 + 16); i++ { - f[i] = byte(2) - } - for i := (8 + 32 + 16); i < (8 + 32 + 16 + 4); i++ { - f[i] = 0xee - } - for i := (8 + 32 + 16 + 4); i < (8 + 32 + 16 + 4 + 2); i++ { - f[i] = byte(3) - } - - return f -} diff --git a/_obfuscated2/obfuscated2.go b/_obfuscated2/obfuscated2.go deleted file mode 100644 index 5fce5dc..0000000 --- a/_obfuscated2/obfuscated2.go +++ /dev/null @@ -1,81 +0,0 @@ -package obfuscated2 - -import ( - "crypto/aes" - "crypto/cipher" - "crypto/sha256" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/mtproto" -) - -// Obfuscated2 contains AES CTR encryption and decryption streams -// for telegram connection. -type Obfuscated2 struct { - Decryptor cipher.Stream - Encryptor cipher.Stream -} - -// ParseObfuscated2ClientFrame parses client frame. Please check this link for -// details: http://telegra.ph/telegram-blocks-wtf-05-26 -// -// Beware, link above is in russian. -func ParseObfuscated2ClientFrame(secret []byte, frame Frame) (*Obfuscated2, *mtproto.ConnectionOpts, error) { - decHasher := sha256.New() - decHasher.Write(frame.Key()) // nolint: errcheck, gosec - decHasher.Write(secret) // nolint: errcheck, gosec - decryptor := makeStreamCipher(decHasher.Sum(nil), frame.IV()) - - invertedFrame := frame.Invert() - encHasher := sha256.New() - encHasher.Write(invertedFrame.Key()) // nolint: errcheck, gosec - encHasher.Write(secret) // nolint: errcheck, gosec - encryptor := makeStreamCipher(encHasher.Sum(nil), invertedFrame.IV()) - - decryptedFrame := make(Frame, FrameLen) - decryptor.XORKeyStream(decryptedFrame, frame) - connType, err := decryptedFrame.ConnectionType() - if err != nil { - return nil, nil, errors.Annotate(err, "Unknown protocol") - } - - obfs := &Obfuscated2{ - Decryptor: decryptor, - Encryptor: encryptor, - } - connOpts := &mtproto.ConnectionOpts{ - DC: decryptedFrame.DC(), - ConnectionType: connType, - } - - return obfs, connOpts, nil -} - -// MakeTelegramObfuscated2Frame creates new handshake frame to send to -// Telegram. -// https://blog.susanka.eu/how-telegram-obfuscates-its-mtproto-traffic/ -func MakeTelegramObfuscated2Frame(opts *mtproto.ConnectionOpts) (*Obfuscated2, Frame) { - frame := generateFrame(opts.ConnectionType) - - encryptor := makeStreamCipher(frame.Key(), frame.IV()) - decryptorFrame := frame.Invert() - decryptor := makeStreamCipher(decryptorFrame.Key(), decryptorFrame.IV()) - - copyFrame := make(Frame, FrameLen) - copy(copyFrame[:frameOffsetIV], frame[:frameOffsetIV]) - encryptor.XORKeyStream(frame, frame) - copy(frame[:frameOffsetIV], copyFrame[:frameOffsetIV]) - - obfs := &Obfuscated2{ - Decryptor: decryptor, - Encryptor: encryptor, - } - - return obfs, frame -} - -func makeStreamCipher(key, iv []byte) cipher.Stream { - block, _ := aes.NewCipher(key) // nolint: gosec - return cipher.NewCTR(block, iv) -} diff --git a/_obfuscated2/obfuscated2_test.go b/_obfuscated2/obfuscated2_test.go deleted file mode 100644 index 7ebe92f..0000000 --- a/_obfuscated2/obfuscated2_test.go +++ /dev/null @@ -1,97 +0,0 @@ -package obfuscated2 - -import ( - "crypto/sha256" - "testing" - - "github.com/stretchr/testify/assert" - - "github.com/9seconds/mtg/mtproto" -) - -func TestObfs2TelegramFrameDecrypt(t *testing.T) { - connOpts := &mtproto.ConnectionOpts{ - DC: 1, - ConnectionType: mtproto.ConnectionTypeIntermediate, - } - _, frame := MakeTelegramObfuscated2Frame(connOpts) - decryptor := makeStreamCipher(frame.Key(), frame.IV()) - - decrypted := make(Frame, FrameLen) - decryptor.XORKeyStream(decrypted, frame) - - _, err := decrypted.ConnectionType() - assert.Nil(t, err) -} - -func TestObfs2TelegramDecryptEncryptDecrypt(t *testing.T) { - connOpts := &mtproto.ConnectionOpts{ - DC: 1, - ConnectionType: mtproto.ConnectionTypeIntermediate, - } - obfs2, frame := MakeTelegramObfuscated2Frame(connOpts) - inverted := frame.Invert() - encryptor := makeStreamCipher(inverted.Key(), inverted.IV()) - - data := []byte{1, 2, 3} - encrypted := make([]byte, 3) - encryptor.XORKeyStream(encrypted, data) - decrypted := make([]byte, 3) - obfs2.Decryptor.XORKeyStream(decrypted, encrypted) - - assert.Equal(t, data, decrypted) -} - -func TestObfs2Full(t *testing.T) { - secret := []byte{1, 2, 3, 4, 5} - - clientFrame := generateFrame(mtproto.ConnectionTypeIntermediate) - clientHasher := sha256.New() - clientHasher.Write(clientFrame.Key()) // nolint: errcheck, gosec - clientHasher.Write(secret) // nolint: errcheck, gosec - clientKey := clientHasher.Sum(nil) - - encryptor := makeStreamCipher(clientKey, clientFrame.IV()) - encrypted := make(Frame, FrameLen) - encryptor.XORKeyStream(encrypted, clientFrame) - copy(encrypted[:56], clientFrame[:56]) - - invertedClientFrame := clientFrame.Invert() - clientHasher = sha256.New() - clientHasher.Write(invertedClientFrame.Key()) // nolint: errcheck, gosec - clientHasher.Write(secret) // nolint: errcheck, gosec - invertedClientKey := clientHasher.Sum(nil) - clientDecryptor := makeStreamCipher(invertedClientKey, invertedClientFrame.IV()) - - clientObfs, _, err := ParseObfuscated2ClientFrame(secret, encrypted) - assert.Nil(t, err) - - connOpts := &mtproto.ConnectionOpts{ - DC: 1, - ConnectionType: mtproto.ConnectionTypeIntermediate, - } - tgObfs, tgFrame := MakeTelegramObfuscated2Frame(connOpts) - tgDecryptor := makeStreamCipher(tgFrame.Key(), tgFrame.IV()) - decrypted := make(Frame, FrameLen) - tgDecryptor.XORKeyStream(decrypted, tgFrame) - _, err = decrypted.ConnectionType() - assert.Nil(t, err) - - tgInvertedFrame := tgFrame.Invert() - tgEncryptor := makeStreamCipher(tgInvertedFrame.Key(), tgInvertedFrame.IV()) - - message := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9} - tgEncryptedMessage := make([]byte, len(message)) - tgEncryptor.XORKeyStream(tgEncryptedMessage, message) - - tgEncDecryptedMessage := make([]byte, len(tgEncryptedMessage)) - tgObfs.Decryptor.XORKeyStream(tgEncDecryptedMessage, tgEncryptedMessage) - assert.Equal(t, message, tgEncDecryptedMessage) - - clientEncryptedMessage := make([]byte, len(tgEncDecryptedMessage)) - clientObfs.Encryptor.XORKeyStream(clientEncryptedMessage, tgEncDecryptedMessage) - finalMessage := make([]byte, len(clientEncryptedMessage)) - clientDecryptor.XORKeyStream(finalMessage, clientEncryptedMessage) - - assert.Equal(t, finalMessage, message) -} diff --git a/_proxy/proxy.go b/_proxy/proxy.go deleted file mode 100644 index ecda26a..0000000 --- a/_proxy/proxy.go +++ /dev/null @@ -1,178 +0,0 @@ -package proxy - -import ( - "context" - "io" - "net" - "sync" - - "github.com/gofrs/uuid" - "github.com/juju/errors" - "go.uber.org/zap" - - "github.com/9seconds/mtg/antireplay" - "github.com/9seconds/mtg/client" - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/stats" - "github.com/9seconds/mtg/telegram" - "github.com/9seconds/mtg/wrappers" -) - -// Proxy is a core of this program. -type Proxy struct { - antiReplayCache antireplay.Cache - clientInit client.Init - tg telegram.Telegram - conf *config.Config -} - -// Serve runs TCP proxy server. -func (p *Proxy) Serve() error { - lsock, err := net.Listen("tcp", p.conf.BindAddr()) - if err != nil { - return errors.Annotate(err, "Cannot create listen socket") - } - - for { - if conn, err := lsock.Accept(); err != nil { - zap.S().Errorw("Cannot allocate incoming connection", "error", err) - } else { - go p.accept(conn) - } - } -} - -func (p *Proxy) accept(conn net.Conn) { - connID := uuid.Must(uuid.NewV4()).String() - log := zap.S().With("connection_id", connID).Named("main") - ctx, cancel := context.WithCancel(context.Background()) - - defer func() { - cancel() - conn.Close() // nolint: errcheck, gosec - - if err := recover(); err != nil { - stats.NewCrash() - log.Errorw("Crash of accept handler", "error", err) - } - }() - - log.Infow("Client connected", "addr", conn.RemoteAddr()) - - clientConn, opts, err := p.clientInit(ctx, cancel, conn, connID, p.antiReplayCache, p.conf) - if err != nil { - log.Errorw("Cannot initialize client connection", "error", err) - return - } - defer clientConn.(io.Closer).Close() // nolint: errcheck - - if p.conf.SecureOnly && opts.ConnectionType != mtproto.ConnectionTypeSecure { - log.Errorw("Proxy supports only secure connections", "connection_type", opts.ConnectionType) - return - } - - stats.ClientConnected(opts.ConnectionType, clientConn.RemoteAddr()) - defer stats.ClientDisconnected(opts.ConnectionType, clientConn.RemoteAddr()) - - serverConn, err := p.getTelegramConn(ctx, cancel, opts, connID) - if err != nil { - log.Errorw("Cannot initialize server connection", "error", err) - return - } - defer serverConn.(io.Closer).Close() // nolint: errcheck - - go func() { - <-ctx.Done() - serverConn.(io.Closer).Close() // nolint: gosec - clientConn.(io.Closer).Close() // nolint: gosec - }() - - wait := &sync.WaitGroup{} - wait.Add(2) - - if p.conf.UseMiddleProxy() { - clientPacket := clientConn.(wrappers.PacketReadWriteCloser) - serverPacket := serverConn.(wrappers.PacketReadWriteCloser) - go p.middlePipe(clientPacket, serverPacket, wait, &opts.ReadHacks) - p.middlePipe(serverPacket, clientPacket, wait, &opts.WriteHacks) - } else { - clientStream := clientConn.(wrappers.StreamReadWriteCloser) - serverStream := serverConn.(wrappers.StreamReadWriteCloser) - go p.directPipe(clientStream, serverStream, wait, p.conf.ReadBufferSize) - p.directPipe(serverStream, clientStream, wait, p.conf.WriteBufferSize) - } - - wait.Wait() - - log.Infow("Client disconnected", "addr", conn.RemoteAddr()) -} - -func (p *Proxy) getTelegramConn(ctx context.Context, cancel context.CancelFunc, - opts *mtproto.ConnectionOpts, connID string) (wrappers.Wrap, error) { - streamConn, err := p.tg.Dial(ctx, cancel, connID, opts) - if err != nil { - return nil, errors.Annotate(err, "Cannot dial to Telegram") - } - - packetConn, err := p.tg.Init(opts, streamConn) - if err != nil { - return nil, errors.Annotate(err, "Cannot handshake telegram") - } - - return packetConn, nil -} - -func (p *Proxy) middlePipe(src wrappers.PacketReadCloser, dst io.Writer, wait *sync.WaitGroup, hacks *mtproto.Hacks) { - defer wait.Done() - - for { - hacks.SimpleAck = false - hacks.QuickAck = false - - packet, err := src.Read() - if err != nil { - src.Logger().Warnw("Cannot read packet", "error", err) - return - } - if _, err = dst.Write(packet); err != nil { - src.Logger().Warnw("Cannot write packet", "error", err) - return - } - } -} - -func (p *Proxy) directPipe(src wrappers.StreamReadCloser, dst io.Writer, wait *sync.WaitGroup, bufferSize int) { - defer wait.Done() - - buffer := make([]byte, bufferSize) - if _, err := io.CopyBuffer(dst, src, buffer); err != nil { - src.Logger().Warnw("Cannot pump sockets", "error", err) - } -} - -// NewProxy returns new proxy instance. -func NewProxy(conf *config.Config) (*Proxy, error) { - var clientInit client.Init - var tg telegram.Telegram - - cache, err := antireplay.NewCache(conf) - if err != nil { - return nil, errors.Annotate(err, "Cannot make proxy") - } - - if conf.UseMiddleProxy() { - clientInit = client.MiddleInit - tg = telegram.NewMiddleTelegram(conf) - } else { - clientInit = client.DirectInit - tg = telegram.NewDirectTelegram(conf) - } - - return &Proxy{ - antiReplayCache: cache, - conf: conf, - clientInit: clientInit, - tg: tg, - }, nil -} diff --git a/_stats/channels.go b/_stats/channels.go deleted file mode 100644 index f7317db..0000000 --- a/_stats/channels.go +++ /dev/null @@ -1,76 +0,0 @@ -package stats - -import ( - "net" - - "github.com/9seconds/mtg/mtproto" -) - -const ( - connectionsChanLength = 10 - trafficChanLength = 10 -) - -var ( - crashesChan = make(chan struct{}) - statsChan = make(chan chan<- Stats) - connectionsChan = make(chan connectionData, connectionsChanLength) - trafficChan = make(chan trafficData, trafficChanLength) -) - -type connectionData struct { - connectionType mtproto.ConnectionType - connected bool - addr *net.TCPAddr -} - -type trafficData struct { - traffic int - ingress bool -} - -// NewCrash indicates new crash. -func NewCrash() { - crashesChan <- struct{}{} -} - -// ClientConnected indicates that new client was connected. -func ClientConnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { - connectionsChan <- connectionData{ - connectionType: connectionType, - addr: addr, - connected: true, - } -} - -// ClientDisconnected indicates that client was disconnected. -func ClientDisconnected(connectionType mtproto.ConnectionType, addr *net.TCPAddr) { - connectionsChan <- connectionData{ - connectionType: connectionType, - addr: addr, - connected: false, - } -} - -// IngressTraffic accounts new ingress traffic. -func IngressTraffic(traffic int) { - trafficChan <- trafficData{ - traffic: traffic, - ingress: true, - } -} - -// EgressTraffic accounts new ingress traffic. -func EgressTraffic(traffic int) { - trafficChan <- trafficData{ - traffic: traffic, - ingress: false, - } -} - -// GetStats returns a snapshot of Stats instance. -func GetStats() Stats { - rpcChan := make(chan Stats) - statsChan <- rpcChan - return <-rpcChan -} diff --git a/_stats/init.go b/_stats/init.go deleted file mode 100644 index ba3640f..0000000 --- a/_stats/init.go +++ /dev/null @@ -1,28 +0,0 @@ -package stats - -import ( - "github.com/juju/errors" - - "github.com/9seconds/mtg/config" -) - -// Init initializes stats subsystem. -func Init(conf *config.Config) error { - if conf.StatsD.Enabled { - client, err := newStatsd(conf) - if err != nil { - return errors.Annotate(err, "Cannot initialize statsd client") - } - go client.run() - } - prometheus, err := newPrometheus(conf) - if err != nil { - return errors.Annotate(err, "Cannot initialize prometheus client") - } - go prometheus.run() - - go NewStats(conf).start() - go startServer(conf, prometheus.getHTTPHandler()) - - return nil -} diff --git a/_stats/prometheus.go b/_stats/prometheus.go deleted file mode 100644 index 00257fa..0000000 --- a/_stats/prometheus.go +++ /dev/null @@ -1,91 +0,0 @@ -package stats - -import ( - "net/http" - "time" - - "github.com/juju/errors" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promhttp" - - "github.com/9seconds/mtg/config" -) - -const prometheusPollTime = time.Second - -type prometheusExporter struct { - registry prometheus.Gatherer - - connections *prometheus.GaugeVec - traffic *prometheus.GaugeVec - speed *prometheus.GaugeVec - crashes prometheus.Gauge -} - -func (p *prometheusExporter) run() { - for range time.Tick(prometheusPollTime) { - instance := GetStats() - - p.connections.WithLabelValues("abridged", "v4").Set(float64(instance.Connections.Abridged.IPv4)) - p.connections.WithLabelValues("abridged", "v6").Set(float64(instance.Connections.Abridged.IPv6)) - p.connections.WithLabelValues("intermediate", "v4").Set(float64(instance.Connections.Intermediate.IPv4)) - p.connections.WithLabelValues("intermediate", "v6").Set(float64(instance.Connections.Intermediate.IPv6)) - p.connections.WithLabelValues("secure", "v4").Set(float64(instance.Connections.Secure.IPv4)) - p.connections.WithLabelValues("secure", "v6").Set(float64(instance.Connections.Secure.IPv6)) - p.traffic.WithLabelValues("ingress").Set(float64(instance.Traffic.ingress)) - p.traffic.WithLabelValues("egress").Set(float64(instance.Traffic.egress)) - p.speed.WithLabelValues("ingress").Set(float64(instance.Speed.ingress)) - p.speed.WithLabelValues("egress").Set(float64(instance.Speed.egress)) - p.crashes.Set(float64(instance.Crashes)) - } -} - -func (p *prometheusExporter) getHTTPHandler() http.Handler { - return promhttp.HandlerFor(p.registry, promhttp.HandlerOpts{}) -} - -func newPrometheus(conf *config.Config) (*prometheusExporter, error) { - registry := prometheus.NewRegistry() - - connections := prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Namespace: conf.Prometheus.Prefix, - Name: "connections", - Help: "Current number of connections to the proxy.", - }, []string{"type", "protocol"}) - traffic := prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Namespace: conf.Prometheus.Prefix, - Name: "traffic", - Help: "Traffic passed through the proxy in bytes.", - }, []string{"direction"}) - speed := prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Namespace: conf.Prometheus.Prefix, - Name: "speed", - Help: "Current throughput in bytes per second.", - }, []string{"direction"}) - crashes := prometheus.NewGauge(prometheus.GaugeOpts{ - Namespace: conf.Prometheus.Prefix, - Name: "crashes", - Help: "How many crashes happened.", - }) - - if err := registry.Register(connections); err != nil { - return nil, errors.Annotate(err, "Cannot register connections collector") - } - if err := registry.Register(traffic); err != nil { - return nil, errors.Annotate(err, "cannot register traffic collector") - } - if err := registry.Register(speed); err != nil { - return nil, errors.Annotate(err, "cannot register speed collector") - } - if err := registry.Register(crashes); err != nil { - return nil, errors.Annotate(err, "cannot register crashes collector") - } - - return &prometheusExporter{ - registry: registry, - connections: connections, - traffic: traffic, - speed: speed, - crashes: crashes, - }, nil -} diff --git a/_stats/server.go b/_stats/server.go deleted file mode 100644 index cc5f659..0000000 --- a/_stats/server.go +++ /dev/null @@ -1,40 +0,0 @@ -package stats - -import ( - "encoding/json" - "net/http" - - "go.uber.org/zap" - - "github.com/9seconds/mtg/config" -) - -func startServer(conf *config.Config, prometheusHandler http.Handler) { - log := zap.S().Named("stats") - - http.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - - first, err := json.Marshal(GetStats()) - if err != nil { - log.Errorw("Cannot encode json", "error", err) - http.Error(w, "Internal server error", 500) - return - } - - interim := map[string]interface{}{} - json.Unmarshal(first, &interim) // nolint: errcheck, gosec - - encoder := json.NewEncoder(w) - encoder.SetEscapeHTML(false) - encoder.SetIndent("", " ") - if err = encoder.Encode(interim); err != nil { - log.Errorw("Cannot encode json", "error", err) - } - }) - http.Handle("/prometheus/", prometheusHandler) - - if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil { - log.Fatalw("Stats server has been stopped", "error", err) - } -} diff --git a/_stats/stats.go b/_stats/stats.go deleted file mode 100644 index 686163d..0000000 --- a/_stats/stats.go +++ /dev/null @@ -1,175 +0,0 @@ -package stats - -import ( - "encoding/json" - "fmt" - "strconv" - "time" - - humanize "github.com/dustin/go-humanize" - - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" -) - -type uptime time.Time - -func (u uptime) MarshalJSON() ([]byte, error) { - duration := time.Since(time.Time(u)) - value := map[string]string{ - "seconds": strconv.Itoa(int(duration.Seconds())), - "human": humanize.Time(time.Time(u)), - } - - return json.Marshal(value) -} - -type connectionType struct { - IPv6 uint32 `json:"ipv6"` - IPv4 uint32 `json:"ipv4"` -} - -type baseConnections struct { - All connectionType `json:"all"` - Abridged connectionType `json:"abridged"` - Intermediate connectionType `json:"intermediate"` - Secure connectionType `json:"secure"` -} - -type connections struct { - baseConnections -} - -func (c connections) MarshalJSON() ([]byte, error) { - c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4 + c.Secure.IPv4 - c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6 + c.Secure.IPv6 - - return json.Marshal(c.baseConnections) -} - -type traffic struct { - ingress uint64 - egress uint64 -} - -func (t *traffic) dumpValue(value uint64) map[string]interface{} { - return map[string]interface{}{ - "bytes": value, - "human": humanize.Bytes(value), - } -} - -func (t traffic) MarshalJSON() ([]byte, error) { - value := map[string]map[string]interface{}{ - "ingress": t.dumpValue(t.ingress), - "egress": t.dumpValue(t.egress), - } - - return json.Marshal(value) -} - -type speed struct { - ingress uint64 - egress uint64 -} - -func (s *speed) dumpValue(value uint64) map[string]interface{} { - return map[string]interface{}{ - "bytes/s": value, - "human": fmt.Sprintf("%s/s", humanize.Bytes(value)), - } -} - -func (s speed) MarshalJSON() ([]byte, error) { - value := map[string]map[string]interface{}{ - "ingress": s.dumpValue(s.ingress), - "egress": s.dumpValue(s.egress), - } - - return json.Marshal(value) -} - -// Stats represents a statistics of the proxy. -type Stats struct { - URLs config.IPURLs `json:"urls"` - Connections connections `json:"connections"` - Traffic traffic `json:"traffic"` - Speed speed `json:"speed"` - Uptime uptime `json:"uptime"` - Crashes uint32 `json:"crashes"` - - previousTraffic traffic -} - -func (s *Stats) start() { - speedChan := time.Tick(time.Second) - - for { - select { - case <-speedChan: - s.handleSpeed() - case event := <-trafficChan: - s.handleTraffic(event) - case event := <-connectionsChan: - s.handleConnection(event) - case getStatsChan := <-statsChan: - s.handleGetStats(getStatsChan) - case <-crashesChan: - s.handleCrash() - } - } -} - -func (s *Stats) handleTraffic(evt trafficData) { - if evt.ingress { - s.Traffic.ingress += uint64(evt.traffic) - } else { - s.Traffic.egress += uint64(evt.traffic) - } -} - -func (s *Stats) handleSpeed() { - s.Speed.ingress = s.Traffic.ingress - s.previousTraffic.ingress - s.Speed.egress = s.Traffic.egress - s.previousTraffic.egress - s.previousTraffic.ingress = s.Traffic.ingress - s.previousTraffic.egress = s.Traffic.egress -} - -func (s *Stats) handleConnection(evt connectionData) { - var inc uint32 = 1 - if !evt.connected { - inc = ^uint32(0) - } - - var conn *connectionType - switch evt.connectionType { - case mtproto.ConnectionTypeAbridged: - conn = &s.Connections.Abridged - case mtproto.ConnectionTypeSecure: - conn = &s.Connections.Secure - default: - conn = &s.Connections.Intermediate - } - - if evt.addr.IP.To4() != nil { - conn.IPv4 += inc - } else { - conn.IPv6 += inc - } -} - -func (s *Stats) handleGetStats(getStatsChan chan<- Stats) { - getStatsChan <- *s -} - -func (s *Stats) handleCrash() { - s.Crashes++ -} - -// NewStats creates a new instance of Stats structure. -func NewStats(conf *config.Config) *Stats { - return &Stats{ - URLs: conf.GetURLs(), - Uptime: uptime(time.Now()), - } -} diff --git a/_stats/statsd.go b/_stats/statsd.go deleted file mode 100644 index 5745889..0000000 --- a/_stats/statsd.go +++ /dev/null @@ -1,77 +0,0 @@ -package stats - -import ( - "time" - - "github.com/juju/errors" - statsd "gopkg.in/alexcesaro/statsd.v2" - - "github.com/9seconds/mtg/config" -) - -const ( - statsdConnectionsAbridgedV4 = "connections.abridged.ipv4" - statsdConnectionsAbridgedV6 = "connections.abridged.ipv6" - - statsdConnectionsIntermediateV4 = "connections.intermediate.ipv4" - statsdConnectionsIntermediateV6 = "connections.intermediate.ipv6" - - statsdConnectionsSecureV4 = "connections.secure.ipv4" - statsdConnectionsSecureV6 = "connections.secure.ipv6" - - statsdTrafficIngress = "traffic.ingress" - statsdTrafficEgress = "traffic.egress" - - statsdSpeedIngress = "speed.ingress" - statsdSpeedEgress = "speed.egress" - - statsdCrashes = "crashes" -) - -const statsdPollTime = time.Second - -type statsdExporter struct { - client *statsd.Client -} - -func (s *statsdExporter) run() { - for range time.Tick(statsdPollTime) { - instance := GetStats() - - s.client.Gauge(statsdConnectionsAbridgedV4, instance.Connections.Abridged.IPv4) - s.client.Gauge(statsdConnectionsAbridgedV6, instance.Connections.Abridged.IPv6) - s.client.Gauge(statsdConnectionsIntermediateV4, instance.Connections.Intermediate.IPv4) - s.client.Gauge(statsdConnectionsIntermediateV6, instance.Connections.Intermediate.IPv6) - s.client.Gauge(statsdConnectionsSecureV4, instance.Connections.Secure.IPv4) - s.client.Gauge(statsdConnectionsSecureV6, instance.Connections.Secure.IPv6) - s.client.Gauge(statsdTrafficIngress, instance.Traffic.ingress) - s.client.Gauge(statsdTrafficEgress, instance.Traffic.egress) - s.client.Gauge(statsdSpeedIngress, instance.Speed.ingress) - s.client.Gauge(statsdSpeedEgress, instance.Speed.egress) - s.client.Gauge(statsdCrashes, instance.Crashes) - } -} - -func newStatsd(conf *config.Config) (*statsdExporter, error) { - options := []statsd.Option{ - statsd.Network(conf.StatsD.Addr.Network()), - statsd.Address(conf.StatsD.Addr.String()), - statsd.Prefix(conf.StatsD.Prefix), - } - - if conf.StatsD.TagsFormat > 0 { - options = append(options, statsd.TagsFormat(conf.StatsD.TagsFormat)) - tags := make([]string, len(conf.StatsD.Tags)*2) - for k, v := range conf.StatsD.Tags { - tags = append(tags, k, v) - } - options = append(options, statsd.Tags(tags...)) - } - - client, err := statsd.New(options...) - if err != nil { - return nil, errors.Annotate(err, "Cannot create statsd client") - } - - return &statsdExporter{client: client}, nil -} diff --git a/_telegram/dialer.go b/_telegram/dialer.go deleted file mode 100644 index 8d192d6..0000000 --- a/_telegram/dialer.go +++ /dev/null @@ -1,52 +0,0 @@ -package telegram - -import ( - "context" - "net" - "time" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/wrappers" -) - -const telegramDialTimeout = 10 * time.Second - -type tgDialer struct { - net.Dialer - - conf *config.Config -} - -func (t *tgDialer) dial(addr string) (net.Conn, error) { - conn, err := t.Dialer.Dial("tcp", addr) - if err != nil { - return nil, errors.Annotate(err, "Cannot connect to Telegram") - } - - tcpSocket := conn.(*net.TCPConn) - if err = tcpSocket.SetNoDelay(true); err != nil { - return nil, errors.Annotate(err, "Cannot set NO_DELAY to Telegram") - } - if err = tcpSocket.SetReadBuffer(t.conf.WriteBufferSize); err != nil { - return nil, errors.Annotate(err, "Cannot set read buffer size on telegram socket") - } - if err = tcpSocket.SetWriteBuffer(t.conf.ReadBufferSize); err != nil { - return nil, errors.Annotate(err, "Cannot set write buffer size on telegram socket") - } - - return conn, nil -} - -func (t *tgDialer) dialRWC(ctx context.Context, cancel context.CancelFunc, - addr, connID string) (wrappers.StreamReadWriteCloser, error) { - conn, err := t.dial(addr) - if err != nil { - return nil, err - } - tgConn := wrappers.NewConn(ctx, cancel, conn, connID, - wrappers.ConnPurposeTelegram, t.conf.PublicIPv4, t.conf.PublicIPv6) - - return tgConn, nil -} diff --git a/_telegram/direct.go b/_telegram/direct.go deleted file mode 100644 index f8a083a..0000000 --- a/_telegram/direct.go +++ /dev/null @@ -1,79 +0,0 @@ -package telegram - -import ( - "context" - "net" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/obfuscated2" - "github.com/9seconds/mtg/wrappers" -) - -const ( - directV4DefaultIdx = 1 - directV6DefaultIdx = 1 -) - -var ( - directV4Addresses = map[int16][]string{ - 0: {"149.154.175.50:443"}, - 1: {"149.154.167.51:443"}, - 2: {"149.154.175.100:443"}, - 3: {"149.154.167.91:443"}, - 4: {"149.154.171.5:443"}, - } - directV6Addresses = map[int16][]string{ - 0: {"[2001:b28:f23d:f001::a]:443"}, - 1: {"[2001:67c:04e8:f002::a]:443"}, - 2: {"[2001:b28:f23d:f003::a]:443"}, - 3: {"[2001:67c:04e8:f004::a]:443"}, - 4: {"[2001:b28:f23f:f005::a]:443"}, - } -) - -type directTelegram struct { - baseTelegram -} - -func (t *directTelegram) Dial(ctx context.Context, cancel context.CancelFunc, - connID string, connOpts *mtproto.ConnectionOpts) (wrappers.StreamReadWriteCloser, error) { - dc := connOpts.DC - if dc < 0 { - dc = -dc - } else if dc == 0 { - dc = 1 - } - - return t.baseTelegram.dial(ctx, cancel, dc-1, connID, connOpts.ConnectionProto) -} - -func (t *directTelegram) Init(connOpts *mtproto.ConnectionOpts, - conn wrappers.StreamReadWriteCloser) (wrappers.Wrap, error) { - obfs2, frame := obfuscated2.MakeTelegramObfuscated2Frame(connOpts) - - if _, err := conn.Write(frame); err != nil { - return nil, errors.Annotate(err, "Cannot write hadnshake frame") - } - - return wrappers.NewStreamCipher(conn, obfs2.Encryptor, obfs2.Decryptor), nil -} - -// NewDirectTelegram returns Telegram instance which connects directly -// to Telegram bypassing middleproxies. -func NewDirectTelegram(conf *config.Config) Telegram { - return &directTelegram{ - baseTelegram: baseTelegram{ - dialer: tgDialer{ - Dialer: net.Dialer{Timeout: telegramDialTimeout}, - conf: conf, - }, - v4DefaultIdx: directV4DefaultIdx, - v6DefaultIdx: directV6DefaultIdx, - v4Addresses: directV4Addresses, - v6Addresses: directV6Addresses, - }, - } -} diff --git a/_telegram/middle.go b/_telegram/middle.go deleted file mode 100644 index 62ff4ba..0000000 --- a/_telegram/middle.go +++ /dev/null @@ -1,139 +0,0 @@ -package telegram - -import ( - "io" - "net" - "net/http" - "sync" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/config" - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/mtproto/rpc" - "github.com/9seconds/mtg/wrappers" -) - -type middleTelegram struct { - middleTelegramCaller - - conf *config.Config -} - -func (t *middleTelegram) Init(connOpts *mtproto.ConnectionOpts, - conn wrappers.StreamReadWriteCloser) (wrappers.Wrap, error) { - rpcNonceConn := wrappers.NewMTProtoFrame(conn, rpc.SeqNoNonce) - - rpcNonceReq, err := t.sendRPCNonceRequest(rpcNonceConn) - if err != nil { - return nil, err - } - rpcNonceResp, err := t.receiveRPCNonceResponse(rpcNonceConn, rpcNonceReq) - if err != nil { - return nil, err - } - - secureConn := wrappers.NewMiddleProxyCipher(conn, rpcNonceReq, rpcNonceResp, t.proxySecret) - frameConn := wrappers.NewMTProtoFrame(secureConn, rpc.SeqNoHandshake) - - rpcHandshakeReq, err := t.sendRPCHandshakeRequest(frameConn) - if err != nil { - return nil, err - } - _, err = t.receiveRPCHandshakeResponse(frameConn, rpcHandshakeReq) - if err != nil { - return nil, err - } - - proxyConn, err := wrappers.NewMTProtoProxy(frameConn, connOpts, t.conf.AdTag) - if err != nil { - return nil, err - } - proxyConn.Logger().Infow("Telegram connection initialized") - - return proxyConn, nil -} - -func (t *middleTelegram) sendRPCNonceRequest(conn io.Writer) (*rpc.NonceRequest, error) { - rpcNonceReq, err := rpc.NewNonceRequest(t.proxySecret) - if err != nil { - return nil, errors.Annotate(err, "Cannot create RPC nonce request") - } - if _, err = conn.Write(rpcNonceReq.Bytes()); err != nil { - return nil, errors.Annotate(err, "Cannot send RPC nonce request") - } - - return rpcNonceReq, nil -} - -func (t *middleTelegram) receiveRPCNonceResponse(conn wrappers.PacketReader, - req *rpc.NonceRequest) (*rpc.NonceResponse, error) { - packet, err := conn.Read() - if err != nil { - return nil, errors.Annotate(err, "Cannot read RPC nonce response") - } - - rpcNonceResp, err := rpc.NewNonceResponse(packet) - if err != nil { - return nil, errors.Annotate(err, "Cannot initialize RPC nonce response") - } - if err = rpcNonceResp.Valid(req); err != nil { - return nil, errors.Annotate(err, "Invalid RPC nonce response") - } - - return rpcNonceResp, nil -} - -func (t *middleTelegram) sendRPCHandshakeRequest(conn io.Writer) (*rpc.HandshakeRequest, error) { - req := rpc.NewHandshakeRequest() - if _, err := conn.Write(req.Bytes()); err != nil { - return nil, errors.Annotate(err, "Cannot send RPC handshake request") - } - - return req, nil -} - -func (t *middleTelegram) receiveRPCHandshakeResponse(conn wrappers.PacketReader, - req *rpc.HandshakeRequest) (*rpc.HandshakeResponse, error) { - packet, err := conn.Read() - if err != nil { - return nil, errors.Annotate(err, "Cannot read RPC handshake response") - } - - rpcHandshakeResp, err := rpc.NewHandshakeResponse(packet) - if err != nil { - return nil, errors.Annotate(err, "Cannot initialize RPC handshake response") - } - if err = rpcHandshakeResp.Valid(req); err != nil { - return nil, errors.Annotate(err, "Invalid RPC handshake response") - } - - return rpcHandshakeResp, nil -} - -// NewMiddleTelegram creates new instance of Telegram which works with -// middle proxies. -func NewMiddleTelegram(conf *config.Config) Telegram { - tg := &middleTelegram{ - middleTelegramCaller: middleTelegramCaller{ - baseTelegram: baseTelegram{ - dialer: tgDialer{ - Dialer: net.Dialer{Timeout: telegramDialTimeout}, - conf: conf, - }, - }, - httpClient: &http.Client{ - Timeout: middleTelegramHTTPClientTimeout, - }, - dialerMutex: &sync.RWMutex{}, - }, - conf: conf, - } - - if err := tg.update(); err != nil { - panic(err) - } - go tg.autoUpdate() - - return tg -} diff --git a/_telegram/middle_caller.go b/_telegram/middle_caller.go deleted file mode 100644 index 47f0780..0000000 --- a/_telegram/middle_caller.go +++ /dev/null @@ -1,191 +0,0 @@ -package telegram - -import ( - "bufio" - "context" - "io/ioutil" - "net" - "net/http" - "regexp" - "strconv" - "strings" - "sync" - "time" - - "github.com/juju/errors" - "go.uber.org/zap" - - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/wrappers" -) - -const ( - middleTelegramAutoUpdateInterval = 6 * time.Hour - middleTelegramHTTPClientTimeout = 30 * time.Second - - tgAddrProxySecret = "https://core.telegram.org/getProxySecret" // nolint: gas - tgAddrProxyV4 = "https://core.telegram.org/getProxyConfig" // nolint: gas - tgAddrProxyV6 = "https://core.telegram.org/getProxyConfigV6" // nolint: gas - tgUserAgent = "mtg" -) - -var middleTelegramProxyConfigSplitter = regexp.MustCompile(`\s+`) - -type middleTelegramCaller struct { - baseTelegram - - proxySecret []byte - dialerMutex *sync.RWMutex - httpClient *http.Client -} - -func (t *middleTelegramCaller) Dial(ctx context.Context, cancel context.CancelFunc, connID string, - connOpts *mtproto.ConnectionOpts) (wrappers.StreamReadWriteCloser, error) { - dc := connOpts.DC - if dc == 0 { - dc = 1 - } - t.dialerMutex.RLock() - defer t.dialerMutex.RUnlock() - - return t.baseTelegram.dial(ctx, cancel, dc, connID, connOpts.ConnectionProto) -} - -func (t *middleTelegramCaller) autoUpdate() { - for range time.Tick(middleTelegramAutoUpdateInterval) { - if err := t.update(); err != nil { - zap.S().Warnw("Cannot update from Telegram", "error", err) - } - } -} - -func (t *middleTelegramCaller) update() error { - secret, err := t.getTelegramProxySecret() - if err != nil { - return errors.Annotate(err, "Cannot get proxy secret") - } - - v4Addresses, v4DefaultIdx, err := t.getTelegramAddresses(tgAddrProxyV4) - if err != nil { - return errors.Annotate(err, "Cannot get ipv4 addresses") - } - - v6Addresses, v6DefaultIdx, err := t.getTelegramAddresses(tgAddrProxyV6) - if err != nil { - return errors.Annotate(err, "Cannot get ipv6 addresses") - } - - t.dialerMutex.Lock() - t.proxySecret = secret - t.v4DefaultIdx = v4DefaultIdx - t.v6DefaultIdx = v6DefaultIdx - t.v4Addresses = v4Addresses - t.v6Addresses = v6Addresses - t.dialerMutex.Unlock() - - zap.S().Infow("Telegram middle proxy data has been updated") - - return nil -} - -func (t *middleTelegramCaller) getTelegramProxySecret() ([]byte, error) { - resp, err := t.call(tgAddrProxySecret) - if err != nil { - return nil, errors.Annotate(err, "Cannot access telegram server") - } - defer resp.Body.Close() // nolint: errcheck - - secret, err := ioutil.ReadAll(resp.Body) - if err != nil { - return nil, errors.Annotate(err, "Cannot read response") - } - - return secret, nil -} - -func (t *middleTelegramCaller) getTelegramAddresses(url string) (map[int16][]string, int16, error) { // nolint: gocyclo - resp, err := t.call(url) - if err != nil { - return nil, 0, errors.Annotate(err, "Cannot access telegram server") - } - defer resp.Body.Close() // nolint: errcheck - - scanner := bufio.NewScanner(resp.Body) - data := map[int16][]string{} - - var defaultIdx int16 = 1 - for scanner.Scan() { - text := strings.TrimSpace(scanner.Text()) - switch { - case strings.HasPrefix(text, "#"): - continue - case strings.HasPrefix(text, "proxy_for"): - addr, idx, err2 := t.parseProxyFor(text) - if err2 != nil { - return nil, 0, errors.Annotate(err2, "Cannot parse 'proxy_for' section") - } - if addresses, ok := data[idx]; ok { - data[idx] = append(addresses, addr) - } else { - data[idx] = []string{addr} - } - case strings.HasPrefix(text, "default"): - idx, err2 := t.parseDefault(text) - if err2 != nil { - return nil, 0, errors.Annotate(err2, "Cannot parse 'default' section") - } - defaultIdx = idx - default: - return nil, 0, errors.Errorf("Unknown config string '%s'", text) - } - } - - err = scanner.Err() - if err != nil { - return nil, 0, errors.Annotate(err, "Cannot read response from the telegram") - } - - return data, defaultIdx, nil -} - -func (t *middleTelegramCaller) parseProxyFor(text string) (string, int16, error) { - chunks := middleTelegramProxyConfigSplitter.Split(text, 3) - if len(chunks) != 3 || chunks[0] != "proxy_for" { - return "", 0, errors.Errorf("Incorrect config '%s'", text) - } - - dcIdx, err := strconv.ParseInt(chunks[1], 10, 16) - if err != nil { - return "", 0, errors.Annotatef(err, "Incorrect config '%s'", text) - } - - addr := strings.TrimRight(chunks[2], ";") - if _, _, err = net.SplitHostPort(addr); err != nil { - return "", 0, errors.Annotatef(err, "Incorrect config '%s'", text) - } - - return addr, int16(dcIdx), nil -} - -func (t *middleTelegramCaller) parseDefault(text string) (int16, error) { - chunks := middleTelegramProxyConfigSplitter.Split(text, 2) - if len(chunks) != 2 || chunks[0] != "default" { - return 0, errors.Errorf("Incorrect config '%s'", text) - } - - dcIdxString := strings.TrimRight(chunks[1], ";") - dcIdx, err := strconv.ParseInt(dcIdxString, 10, 16) - if err != nil { - return 0, errors.Annotatef(err, "Incorrect config '%s'", text) - } - - return int16(dcIdx), nil -} - -func (t *middleTelegramCaller) call(url string) (*http.Response, error) { - req, _ := http.NewRequest("GET", url, nil) // nolint: gosec - req.Header.Set("Accept", "text/plain") - req.Header.Set("User-Agent", tgUserAgent) - - return t.httpClient.Do(req) -} diff --git a/_telegram/telegram.go b/_telegram/telegram.go deleted file mode 100644 index c7fd41e..0000000 --- a/_telegram/telegram.go +++ /dev/null @@ -1,68 +0,0 @@ -package telegram - -import ( - "context" - "math/rand" - - "github.com/juju/errors" - - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/wrappers" -) - -// Telegram is an interface for different Telegram work modes. -type Telegram interface { - Dial(context.Context, context.CancelFunc, string, *mtproto.ConnectionOpts) (wrappers.StreamReadWriteCloser, error) - Init(*mtproto.ConnectionOpts, wrappers.StreamReadWriteCloser) (wrappers.Wrap, error) -} - -type baseTelegram struct { - dialer tgDialer - - v4DefaultIdx int16 - v6DefaultIdx int16 - v4Addresses map[int16][]string - v6Addresses map[int16][]string -} - -func (b *baseTelegram) dial(ctx context.Context, cancel context.CancelFunc, dcIdx int16, connID string, - proto mtproto.ConnectionProtocol) (wrappers.StreamReadWriteCloser, error) { - addrs := make([]string, 2) - - if proto&mtproto.ConnectionProtocolIPv6 != 0 { - if addr := b.chooseAddress(b.v6Addresses, dcIdx, b.v6DefaultIdx); addr != "" { - addrs = append(addrs, addr) - } - } - if proto&mtproto.ConnectionProtocolIPv4 != 0 { - if addr := b.chooseAddress(b.v4Addresses, dcIdx, b.v4DefaultIdx); addr != "" { - addrs = append(addrs, addr) - } - } - - for _, addr := range addrs { - if conn, err := b.dialer.dialRWC(ctx, cancel, addr, connID); err == nil { - return conn, err - } - } - - return nil, errors.New("Cannot connect to Telegram") -} - -func (b *baseTelegram) chooseAddress(addresses map[int16][]string, idx, defaultIdx int16) string { - if addr, ok := addresses[idx]; ok { - return b.chooseRandomAddress(addr) - } else if addr, ok := addresses[defaultIdx]; ok { - return b.chooseRandomAddress(addr) - } - - return "" -} - -func (b *baseTelegram) chooseRandomAddress(addresses []string) string { - if len(addresses) > 0 { - return addresses[rand.Intn(len(addresses))] - } - - return "" -} diff --git a/_utils/read_current_data.go b/_utils/read_current_data.go deleted file mode 100644 index 284369c..0000000 --- a/_utils/read_current_data.go +++ /dev/null @@ -1,21 +0,0 @@ -package utils - -import "io" - -const readCurrentDataBufferSize = 1024 + 1 // + 1 because telegram operates with blocks mod 4 - -// ReadCurrentData reads all data from io.Reader which is ready to be read. -func ReadCurrentData(src io.Reader) (rv []byte, err error) { - buf := make([]byte, readCurrentDataBufferSize) - n := readCurrentDataBufferSize - - for n == len(buf) { - n, err = src.Read(buf) - if err != nil { - return nil, err - } - rv = append(rv, buf[:n]...) - } - - return rv, nil -} diff --git a/_utils/reverse_bytes.go b/_utils/reverse_bytes.go deleted file mode 100644 index ab7cd5b..0000000 --- a/_utils/reverse_bytes.go +++ /dev/null @@ -1,15 +0,0 @@ -package utils - -// ReverseBytes is a common slice reverser. -func ReverseBytes(data []byte) []byte { - dataLen := len(data) - rv := make([]byte, dataLen) - - rv[dataLen/2] = data[dataLen/2] - for i := dataLen/2 - 1; i >= 0; i-- { - opp := dataLen - i - 1 - rv[i], rv[opp] = data[opp], data[i] - } - - return rv -} diff --git a/_utils/uint24.go b/_utils/uint24.go deleted file mode 100644 index 66c5f1b..0000000 --- a/_utils/uint24.go +++ /dev/null @@ -1,15 +0,0 @@ -package utils - -// Uint24 is a replacement for the absent Go uint24 data type. -// This data type is little endian. -type Uint24 [3]byte - -// ToUint24 converts number to Uint24. -func ToUint24(number uint32) Uint24 { - return Uint24{byte(number), byte(number >> 8), byte(number >> 16)} -} - -// FromUint24 converts Uint24 to number. -func FromUint24(number Uint24) uint32 { - return uint32(number[0]) + (uint32(number[1]) << 8) + (uint32(number[2]) << 16) -} diff --git a/_wrappers/blockcipher.go b/_wrappers/blockcipher.go deleted file mode 100644 index 745efca..0000000 --- a/_wrappers/blockcipher.go +++ /dev/null @@ -1,99 +0,0 @@ -package wrappers - -import ( - "bytes" - "crypto/aes" - "crypto/cipher" - "net" - - "go.uber.org/zap" - - "github.com/9seconds/mtg/utils" - "github.com/juju/errors" -) - -// BlockCipher is a stream writer which encrypts/decrypts blocks of data -// with AES CBC. This also is buffered reader. It means, that block -// reading is transparent for it, you can assume you are working with -// good old io.Reader. -type BlockCipher struct { - buf *bytes.Buffer - - logger *zap.SugaredLogger - conn StreamReadWriteCloser - encryptor cipher.BlockMode - decryptor cipher.BlockMode -} - -func (b *BlockCipher) Read(p []byte) (int, error) { - if b.buf.Len() > 0 { - return b.flush(p) - } - - buf := []byte{} - for len(buf) == 0 || len(buf)%aes.BlockSize != 0 { - rv, err := utils.ReadCurrentData(b.conn) - if err != nil { - return 0, errors.Annotate(err, "Cannot read from socket") - } - buf = append(buf, rv...) - } - - b.decryptor.CryptBlocks(buf, buf) - b.buf.Write(buf) // nolint: gosec - - return b.flush(p) -} - -func (b *BlockCipher) flush(p []byte) (int, error) { - if b.buf.Len() <= len(p) { - sizeToReturn := b.buf.Len() - copy(p, b.buf.Bytes()) - b.buf.Reset() - return sizeToReturn, nil - } - - return b.buf.Read(p) -} - -func (b *BlockCipher) Write(p []byte) (int, error) { - if len(p)%aes.BlockSize > 0 { - return 0, errors.Errorf("Incorrect block size %d", len(p)) - } - - encrypted := make([]byte, len(p)) - b.encryptor.CryptBlocks(encrypted, p) - - return b.conn.Write(encrypted) -} - -// Logger returns an instance of the logger for this wrapper. -func (b *BlockCipher) Logger() *zap.SugaredLogger { - return b.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (b *BlockCipher) LocalAddr() *net.TCPAddr { - return b.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (b *BlockCipher) RemoteAddr() *net.TCPAddr { - return b.conn.RemoteAddr() -} - -// Close closes underlying net.Conn. -func (b *BlockCipher) Close() error { - return b.conn.Close() -} - -// NewBlockCipher creates new instance of BlockCipher based on given data. -func NewBlockCipher(conn StreamReadWriteCloser, encryptor, decryptor cipher.BlockMode) StreamReadWriteCloser { - return &BlockCipher{ - buf: &bytes.Buffer{}, - conn: conn, - logger: conn.Logger().Named("block-cipher"), - encryptor: encryptor, - decryptor: decryptor, - } -} diff --git a/_wrappers/conn.go b/_wrappers/conn.go deleted file mode 100644 index 239af53..0000000 --- a/_wrappers/conn.go +++ /dev/null @@ -1,154 +0,0 @@ -package wrappers - -import ( - "context" - "net" - "time" - - "go.uber.org/zap" - - "github.com/9seconds/mtg/stats" - "github.com/juju/errors" -) - -// ConnPurpose is intended to be identifier of connection purpose. We -// sometimes want to treat client/telegram connection differently (for -// logging for example). -type ConnPurpose uint8 - -func (c ConnPurpose) String() string { - switch c { - case ConnPurposeClient: - return "client" - case ConnPurposeTelegram: - return "telegram" - } - - return "" -} - -// ConnPurpose* define different connection types. -const ( - ConnPurposeClient = iota - ConnPurposeTelegram -) - -const ( - connTimeoutRead = 2 * time.Minute - connTimeoutWrite = 2 * time.Minute -) - -// Conn is a basic wrapper for net.Conn providing the most low-level -// logic and management as possible. -type Conn struct { - conn net.Conn - ctx context.Context - cancel context.CancelFunc - connID string - logger *zap.SugaredLogger - - publicIPv4 net.IP - publicIPv6 net.IP -} - -func (c *Conn) Write(p []byte) (int, error) { - select { - case <-c.ctx.Done(): - c.Close() // nolint: gosec - return 0, errors.Annotate(c.ctx.Err(), "Cannot write because context was closed") - default: - if err := c.conn.SetWriteDeadline(time.Now().Add(connTimeoutWrite)); err != nil { - c.Close() // nolint: gosec - return 0, errors.Annotate(err, "Cannot set write deadline to the socket") - } - - n, err := c.conn.Write(p) - c.logger.Debugw("Write to stream", "bytes", n, "error", err) - stats.EgressTraffic(n) - if err != nil { - c.Close() // nolint: gosec - } - - return n, err - } -} - -func (c *Conn) Read(p []byte) (int, error) { - select { - case <-c.ctx.Done(): - c.Close() // nolint: gosec - return 0, errors.Annotate(c.ctx.Err(), "Cannot read because context was closed") - default: - if err := c.conn.SetReadDeadline(time.Now().Add(connTimeoutRead)); err != nil { - c.Close() // nolint: gosec - return 0, errors.Annotate(err, "Cannot set read deadline to the socket") - } - - n, err := c.conn.Read(p) - c.logger.Debugw("Read from stream", "bytes", n, "error", err) - stats.IngressTraffic(n) - if err != nil { - c.Close() // nolint: gosec - } - - return n, err - } -} - -// Close closes underlying net.Conn instance. -func (c *Conn) Close() error { - c.logger.Debugw("Close connection") - c.cancel() - - return c.conn.Close() -} - -// Logger returns an instance of the logger for this wrapper. -func (c *Conn) Logger() *zap.SugaredLogger { - return c.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (c *Conn) LocalAddr() *net.TCPAddr { - addr := c.conn.LocalAddr().(*net.TCPAddr) - newAddr := *addr - - if c.RemoteAddr().IP.To4() != nil { - if c.publicIPv4 != nil { - newAddr.IP = c.publicIPv4 - } - } else if c.publicIPv6 != nil { - newAddr.IP = c.publicIPv6 - } - - return &newAddr -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (c *Conn) RemoteAddr() *net.TCPAddr { - return c.conn.RemoteAddr().(*net.TCPAddr) -} - -// NewConn initializes Conn wrapper for net.Conn. -func NewConn(ctx context.Context, cancel context.CancelFunc, conn net.Conn, - connID string, purpose ConnPurpose, publicIPv4, publicIPv6 net.IP) StreamReadWriteCloser { - logger := zap.S().With( - "connection_id", connID, - "local_address", conn.LocalAddr(), - "remote_address", conn.RemoteAddr(), - "purpose", purpose, - ).Named("conn") - - wrapper := Conn{ - conn: conn, - ctx: ctx, - cancel: cancel, - connID: connID, - logger: logger, - publicIPv4: publicIPv4, - publicIPv6: publicIPv6, - } - wrapper.logger = logger.With("faked_local_addr", wrapper.LocalAddr()) - - return &wrapper -} diff --git a/_wrappers/mtproto_abridged.go b/_wrappers/mtproto_abridged.go deleted file mode 100644 index 6332af6..0000000 --- a/_wrappers/mtproto_abridged.go +++ /dev/null @@ -1,159 +0,0 @@ -package wrappers - -import ( - "bytes" - "io" - "net" - - "github.com/juju/errors" - "go.uber.org/zap" - - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/utils" -) - -const ( - mtprotoAbridgedSmallPacketLength = 0x7f - mtprotoAbridgedQuickAckLength = 0x80 - mtprotoAbridgedLargePacketLength = 16777216 // 256 ^ 3 -) - -// MTProtoAbridged presents abridged connection between client and -// middle proxy. -type MTProtoAbridged struct { - conn StreamReadWriteCloser - opts *mtproto.ConnectionOpts - logger *zap.SugaredLogger - - readCounter uint32 - writeCounter uint32 -} - -func (m *MTProtoAbridged) Read() ([]byte, error) { - defer func() { - m.readCounter++ - }() - - m.logger.Debugw("Read packet", - "simple_ack", m.opts.ReadHacks.SimpleAck, - "quick_ack", m.opts.ReadHacks.QuickAck, - "counter", m.readCounter, - ) - - buf := &bytes.Buffer{} - buf.Grow(3) - - if _, err := io.CopyN(buf, m.conn, 1); err != nil { - return nil, errors.Annotate(err, "Cannot read message length") - } - msgLength := uint32(buf.Bytes()[0]) - buf.Reset() - - m.logger.Debugw("Packet first byte", - "byte", msgLength, - "counter", m.readCounter, - "simple_ack", m.opts.ReadHacks.SimpleAck, - "quick_ack", m.opts.ReadHacks.QuickAck, - ) - - if msgLength >= mtprotoAbridgedQuickAckLength { - m.opts.ReadHacks.QuickAck = true - msgLength -= mtprotoAbridgedQuickAckLength - } - - if msgLength == mtprotoAbridgedSmallPacketLength { - if _, err := io.CopyN(buf, m.conn, 3); err != nil { - return nil, errors.Annotate(err, "Cannot read the correct message length") - } - number := utils.Uint24{} - copy(number[:], buf.Bytes()) - msgLength = utils.FromUint24(number) - } - msgLength *= 4 - - m.logger.Debugw("Packet length", - "length", msgLength, - "simple_ack", m.opts.ReadHacks.SimpleAck, - "quick_ack", m.opts.ReadHacks.QuickAck, - "counter", m.readCounter, - ) - - buf.Reset() - buf.Grow(int(msgLength)) - if _, err := io.CopyN(buf, m.conn, int64(msgLength)); err != nil { - return nil, errors.Annotate(err, "Cannot read message") - } - - return buf.Bytes(), nil -} - -func (m *MTProtoAbridged) Write(p []byte) (int, error) { - defer func() { - m.writeCounter++ - }() - - m.logger.Debugw("Write packet", - "length", len(p), - "simple_ack", m.opts.WriteHacks.SimpleAck, - "quick_ack", m.opts.WriteHacks.QuickAck, - "counter", m.writeCounter, - ) - - if len(p)%4 != 0 { - return 0, errors.Errorf("Incorrect packet length %d", len(p)) - } - - if m.opts.WriteHacks.SimpleAck { - return m.conn.Write(utils.ReverseBytes(p)) - } - - packetLength := len(p) / 4 - switch { - case packetLength < mtprotoAbridgedSmallPacketLength: - newData := append([]byte{byte(packetLength)}, p...) - return m.conn.Write(newData) - - case packetLength < mtprotoAbridgedLargePacketLength: - length24 := utils.ToUint24(uint32(packetLength)) - - buf := &bytes.Buffer{} - buf.Grow(1 + 3 + len(p)) - - buf.WriteByte(byte(mtprotoAbridgedSmallPacketLength)) // nolint: gosec - buf.Write(length24[:]) // nolint: gosec - buf.Write(p) // nolint: gosec - - return m.conn.Write(buf.Bytes()) - } - - return 0, errors.Errorf("Packet is too big %d", len(p)) -} - -// Logger returns an instance of the logger for this wrapper. -func (m *MTProtoAbridged) Logger() *zap.SugaredLogger { - return m.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (m *MTProtoAbridged) LocalAddr() *net.TCPAddr { - return m.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (m *MTProtoAbridged) RemoteAddr() *net.TCPAddr { - return m.conn.RemoteAddr() -} - -// Close closes underlying net.Conn instance. -func (m *MTProtoAbridged) Close() error { - return m.conn.Close() -} - -// NewMTProtoAbridged creates new wrapper for abridged client connection. -func NewMTProtoAbridged(conn StreamReadWriteCloser, opts *mtproto.ConnectionOpts) PacketReadWriteCloser { - return &MTProtoAbridged{ - conn: conn, - opts: opts, - logger: conn.Logger().Named("mtproto-abridged"), - } -} diff --git a/_wrappers/mtproto_cipher.go b/_wrappers/mtproto_cipher.go deleted file mode 100644 index 2f65b0b..0000000 --- a/_wrappers/mtproto_cipher.go +++ /dev/null @@ -1,98 +0,0 @@ -package wrappers - -import ( - "bytes" - "crypto/aes" - "crypto/cipher" - "crypto/md5" // nolint: gas - "crypto/sha1" // nolint: gosec - "encoding/binary" - "net" - - "github.com/9seconds/mtg/mtproto/rpc" - "github.com/9seconds/mtg/utils" -) - -type cipherPurpose uint8 - -const ( - cipherPurposeClient cipherPurpose = iota - cipherPurposeServer -) - -var emptyIP = [4]byte{0x00, 0x00, 0x00, 0x00} - -// NewMiddleProxyCipher creates new block cipher to proxy<->telegram -// connection. -func NewMiddleProxyCipher(conn StreamReadWriteCloser, - req *rpc.NonceRequest, resp *rpc.NonceResponse, secret []byte) StreamReadWriteCloser { - localAddr := conn.LocalAddr() - remoteAddr := conn.RemoteAddr() - - encKey, encIV := deriveKeys(cipherPurposeClient, req, resp, localAddr, remoteAddr, secret) - decKey, decIV := deriveKeys(cipherPurposeServer, req, resp, localAddr, remoteAddr, secret) - - enc, _ := makeEncrypterDecrypter(encKey, encIV) - _, dec := makeEncrypterDecrypter(decKey, decIV) - - return NewBlockCipher(conn, enc, dec) -} - -func deriveKeys(purpose cipherPurpose, req *rpc.NonceRequest, resp *rpc.NonceResponse, - client, remote *net.TCPAddr, secret []byte) ([]byte, []byte) { - message := bytes.Buffer{} - message.Write(resp.Nonce) // nolint: gosec - message.Write(req.Nonce) // nolint: gosec - message.Write(req.CryptoTS) // nolint: gosec - - clientIPv4 := emptyIP[:] - serverIPv4 := emptyIP[:] - if client.IP.To4() != nil { - clientIPv4 = utils.ReverseBytes(client.IP.To4()) - serverIPv4 = utils.ReverseBytes(remote.IP.To4()) - } - message.Write(serverIPv4) // nolint: gosec - - var port [2]byte - binary.LittleEndian.PutUint16(port[:], uint16(client.Port)) - message.Write(port[:]) // nolint: gosec - - switch purpose { - case cipherPurposeClient: - message.WriteString("CLIENT") // nolint: gosec - case cipherPurposeServer: - message.WriteString("SERVER") // nolint: gosec - default: - panic("Unexpected cipher purpose") - } - - message.Write(clientIPv4) // nolint: gosec - binary.LittleEndian.PutUint16(port[:], uint16(remote.Port)) - message.Write(port[:]) // nolint: gosec - message.Write(secret) // nolint: gosec - message.Write(resp.Nonce) // nolint: gosec - - if client.IP.To4() == nil { - message.Write(client.IP.To16()) // nolint: gosec - message.Write(remote.IP.To16()) // nolint: gosec - } - message.Write(req.Nonce) // nolint: gosec - - data := message.Bytes() - md5sum := md5.Sum(data[1:]) // nolint: gas - sha1sum := sha1.Sum(data) // nolint: gosec - - key := append(md5sum[:12], sha1sum[:]...) - iv := md5.Sum(data[2:]) // nolint: gas - - return key, iv[:] -} - -func makeEncrypterDecrypter(key, iv []byte) (cipher.BlockMode, cipher.BlockMode) { - block, err := aes.NewCipher(key) - if err != nil { - panic(err) - } - - return cipher.NewCBCEncrypter(block, iv), cipher.NewCBCDecrypter(block, iv) -} diff --git a/_wrappers/mtproto_frame.go b/_wrappers/mtproto_frame.go deleted file mode 100644 index 2e39d6b..0000000 --- a/_wrappers/mtproto_frame.go +++ /dev/null @@ -1,161 +0,0 @@ -package wrappers - -import ( - "bytes" - "crypto/aes" - "encoding/binary" - "hash/crc32" - "io" - "io/ioutil" - "net" - - "github.com/juju/errors" - "go.uber.org/zap" -) - -const ( - mtprotoFrameMinMessageLength = 12 - mtprotoFrameMaxMessageLength = 16777216 -) - -var mtprotoFramePadding = []byte{0x04, 0x00, 0x00, 0x00} - -// MTProtoFrame is a wrapper which converts written data to the MTProtoFrame. -// The format of the frame: -// -// [ MSGLEN(4) | SEQNO(4) | MSG(...) | CRC32(4) | PADDING(4*x) ] -// -// MSGLEN is the length of the message + len of seqno and msglen. -// SEQNO is the number of frame in the receive/send sequence. If client -// sends a message with SeqNo 18, it has to receive message with SeqNo 18. -// MSG is the data which has to be written -// CRC32 is the CRC32 checksum of MSGLEN + SEQNO + MSG -// PADDING is custom padding schema to complete frame length to such that -// len(frame) % 16 == 0 -type MTProtoFrame struct { - conn StreamReadWriteCloser - logger *zap.SugaredLogger - - readSeqNo int32 - writeSeqNo int32 -} - -func (m *MTProtoFrame) Read() ([]byte, error) { // nolint: gocyclo - buf := &bytes.Buffer{} - sum := crc32.NewIEEE() - writer := io.MultiWriter(buf, sum) - - for { - buf.Reset() - sum.Reset() - if _, err := io.CopyN(writer, m.conn, 4); err != nil { - return nil, errors.Annotate(err, "Cannot read frame padding") - } - if !bytes.Equal(buf.Bytes(), mtprotoFramePadding) { - break - } - } - - messageLength := binary.LittleEndian.Uint32(buf.Bytes()) - m.logger.Debugw("Read MTProto frame", - "messageLength", messageLength, - "sequence_number", m.readSeqNo, - ) - if messageLength%4 != 0 || messageLength < mtprotoFrameMinMessageLength || - messageLength > mtprotoFrameMaxMessageLength { - return nil, errors.Errorf("Incorrect frame message length %d", messageLength) - } - - buf.Reset() - buf.Grow(int(messageLength) - 4 - 4) - if _, err := io.CopyN(writer, m.conn, int64(messageLength)-4-4); err != nil { - return nil, errors.Annotate(err, "Cannot read the message frame") - } - - var seqNo int32 - binary.Read(buf, binary.LittleEndian, &seqNo) // nolint: errcheck, gosec - if seqNo != m.readSeqNo { - return nil, errors.Errorf("Unexpected sequence number %d (wait for %d)", seqNo, m.readSeqNo) - } - - data, _ := ioutil.ReadAll(buf) // nolint: gosec - buf.Reset() - // write to buf, not to writer. This is because we are going to fetch - // crc32 checksum. - if _, err := io.CopyN(buf, m.conn, 4); err != nil { - return nil, errors.Annotate(err, "Cannot read checksum") - } - - checksum := binary.LittleEndian.Uint32(buf.Bytes()) - if checksum != sum.Sum32() { - return nil, errors.Errorf("CRC32 checksum mismatch. Wait for %d, got %d", sum.Sum32(), checksum) - } - - m.logger.Debugw("Read MTProto frame", - "messageLength", messageLength, - "sequence_number", m.readSeqNo, - "dataLength", len(data), - "checksum", checksum, - ) - m.readSeqNo++ - - return data, nil -} - -func (m *MTProtoFrame) Write(p []byte) (int, error) { - messageLength := 4 + 4 + len(p) + 4 - paddingLength := (aes.BlockSize - messageLength%aes.BlockSize) % aes.BlockSize - - buf := &bytes.Buffer{} - buf.Grow(messageLength + paddingLength) - - binary.Write(buf, binary.LittleEndian, uint32(messageLength)) // nolint: errcheck, gosec - binary.Write(buf, binary.LittleEndian, m.writeSeqNo) // nolint: errcheck, gosec - buf.Write(p) // nolint: gosec - - checksum := crc32.ChecksumIEEE(buf.Bytes()) - binary.Write(buf, binary.LittleEndian, checksum) // nolint: errcheck, gosec - buf.Write(bytes.Repeat(mtprotoFramePadding, paddingLength/4)) // nolint: gosec - - m.logger.Debugw("Write MTProto frame", - "length", len(p), - "sequence_number", m.writeSeqNo, - "crc32", checksum, - "frame_length", buf.Len(), - ) - m.writeSeqNo++ - - _, err := m.conn.Write(buf.Bytes()) - - return len(p), err -} - -// Logger returns an instance of the logger for this wrapper. -func (m *MTProtoFrame) Logger() *zap.SugaredLogger { - return m.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (m *MTProtoFrame) LocalAddr() *net.TCPAddr { - return m.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (m *MTProtoFrame) RemoteAddr() *net.TCPAddr { - return m.conn.RemoteAddr() -} - -// Close closes underlying net.Conn instance. -func (m *MTProtoFrame) Close() error { - return m.conn.Close() -} - -// NewMTProtoFrame creates new PacketWrapper for underlying connection. -func NewMTProtoFrame(conn StreamReadWriteCloser, seqNo int32) PacketReadWriteCloser { - return &MTProtoFrame{ - conn: conn, - logger: conn.Logger().Named("mtproto-frame"), - readSeqNo: seqNo, - writeSeqNo: seqNo, - } -} diff --git a/_wrappers/mtproto_intermediate.go b/_wrappers/mtproto_intermediate.go deleted file mode 100644 index 5b88d74..0000000 --- a/_wrappers/mtproto_intermediate.go +++ /dev/null @@ -1,117 +0,0 @@ -package wrappers - -import ( - "bytes" - "encoding/binary" - "io" - "net" - - "github.com/juju/errors" - "go.uber.org/zap" - - "github.com/9seconds/mtg/mtproto" -) - -const mtprotoIntermediateQuickAckLength = 0x80000000 - -// MTProtoIntermediate presents intermediate connection between client -// and Telegram. -type MTProtoIntermediate struct { - conn StreamReadWriteCloser - opts *mtproto.ConnectionOpts - logger *zap.SugaredLogger - - readCounter uint32 - writeCounter uint32 -} - -func (m *MTProtoIntermediate) Read() ([]byte, error) { - defer func() { - m.readCounter++ - }() - - m.logger.Debugw("Read packet", - "simple_ack", m.opts.ReadHacks.SimpleAck, - "quick_ack", m.opts.ReadHacks.QuickAck, - "counter", m.readCounter, - ) - - buf := &bytes.Buffer{} - buf.Grow(4) - - if _, err := io.CopyN(buf, m.conn, 4); err != nil { - return nil, errors.Annotate(err, "Cannot read message length") - } - length := binary.LittleEndian.Uint32(buf.Bytes()) - - m.logger.Debugw("Packet message length", - "simple_ack", m.opts.ReadHacks.SimpleAck, - "quick_ack", m.opts.ReadHacks.QuickAck, - "counter", m.readCounter, - "length", length, - ) - - if length > mtprotoIntermediateQuickAckLength { - m.opts.ReadHacks.QuickAck = true - length -= mtprotoIntermediateQuickAckLength - } - - buf.Reset() - buf.Grow(int(length)) - if _, err := io.CopyN(buf, m.conn, int64(length)); err != nil { - return nil, errors.Annotate(err, "Cannot read the message") - } - - return buf.Bytes()[:length], nil -} - -func (m *MTProtoIntermediate) Write(p []byte) (int, error) { - defer func() { - m.writeCounter++ - }() - - m.logger.Debugw("Write packet", - "simple_ack", m.opts.WriteHacks.SimpleAck, - "quick_ack", m.opts.WriteHacks.QuickAck, - "counter", m.writeCounter, - ) - - if m.opts.WriteHacks.SimpleAck { - return m.conn.Write(p) - } - - var length [4]byte - binary.LittleEndian.PutUint32(length[:], uint32(len(p))) - - return m.conn.Write(append(length[:], p...)) -} - -// Logger returns an instance of the logger for this wrapper. -func (m *MTProtoIntermediate) Logger() *zap.SugaredLogger { - return m.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (m *MTProtoIntermediate) LocalAddr() *net.TCPAddr { - return m.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (m *MTProtoIntermediate) RemoteAddr() *net.TCPAddr { - return m.conn.RemoteAddr() -} - -// Close closes underlying net.Conn instance. -func (m *MTProtoIntermediate) Close() error { - return m.conn.Close() -} - -// NewMTProtoIntermediate creates new PacketWrapper for intermediate -// client connection. -func NewMTProtoIntermediate(conn StreamReadWriteCloser, opts *mtproto.ConnectionOpts) PacketReadWriteCloser { - return &MTProtoIntermediate{ - conn: conn, - logger: conn.Logger().Named("mtproto-intermediate"), - opts: opts, - } -} diff --git a/_wrappers/mtproto_intermediate_secure.go b/_wrappers/mtproto_intermediate_secure.go deleted file mode 100644 index f26612b..0000000 --- a/_wrappers/mtproto_intermediate_secure.go +++ /dev/null @@ -1,74 +0,0 @@ -package wrappers - -import ( - "bytes" - "encoding/binary" - "math/rand" - - "github.com/9seconds/mtg/mtproto" -) - -// MTProtoIntermediateSecure is an extension of MTProtoIntermediate -// mode which supports random paddings (socalled 'secure mode' or -// 'dd-secrets'). -type MTProtoIntermediateSecure struct { - MTProtoIntermediate -} - -func (m *MTProtoIntermediateSecure) Read() ([]byte, error) { - data, err := m.MTProtoIntermediate.Read() - if err != nil { - return nil, err - } - length := len(data) - (len(data) % 4) - - return data[:length], nil -} - -func (m *MTProtoIntermediateSecure) Write(p []byte) (int, error) { - defer func() { - m.writeCounter++ - }() - - m.logger.Debugw("Write packet", - "simple_ack", m.opts.WriteHacks.SimpleAck, - "quick_ack", m.opts.WriteHacks.QuickAck, - "counter", m.writeCounter, - ) - - if m.opts.WriteHacks.SimpleAck { - return m.conn.Write(p) - } - - buf := &bytes.Buffer{} - paddingLength := rand.Intn(4) - buf.Grow(4 + len(p) + paddingLength) - - binary.Write(buf, binary.LittleEndian, uint32(len(p)+paddingLength)) // nolint: errcheck, gosec - buf.Write(p) // nolint: gosec - buf.Write(make([]byte, paddingLength)) // nolint: gosec - - m.logger.Debugw("Write packet with padding", - "simple_ack", m.opts.WriteHacks.SimpleAck, - "quick_ack", m.opts.WriteHacks.QuickAck, - "counter", m.writeCounter, - "padding_length", paddingLength, - "length", len(p), - ) - - _, err := m.conn.Write(buf.Bytes()) - - return len(p), err -} - -// NewMTProtoIntermediateSecure create new instance of -// MTProtoIntermediateSecure instance. -func NewMTProtoIntermediateSecure(conn StreamReadWriteCloser, opts *mtproto.ConnectionOpts) PacketReadWriteCloser { - return &MTProtoIntermediateSecure{ - MTProtoIntermediate: MTProtoIntermediate{ - conn: conn, - logger: conn.Logger().Named("mtproto-intermediate-secure"), - opts: opts, - }, - } -} diff --git a/_wrappers/mtproto_proxy.go b/_wrappers/mtproto_proxy.go deleted file mode 100644 index e3ac897..0000000 --- a/_wrappers/mtproto_proxy.go +++ /dev/null @@ -1,165 +0,0 @@ -package wrappers - -import ( - "bytes" - "fmt" - "net" - - "github.com/juju/errors" - "go.uber.org/zap" - - "github.com/9seconds/mtg/mtproto" - "github.com/9seconds/mtg/mtproto/rpc" -) - -// MTProtoProxy is a wrapper which creates/reads RPC responses from Telegram. -type MTProtoProxy struct { - conn PacketReadWriteCloser - req *rpc.ProxyRequest - logger *zap.SugaredLogger - - readCounter uint32 - writeCounter uint32 -} - -func (m *MTProtoProxy) Read() ([]byte, error) { - defer func() { - m.readCounter++ - }() - - m.logger.Debugw("Read packet", - "counter", m.readCounter, - "simple_ack", m.req.Options.WriteHacks.SimpleAck, - "quick_ack", m.req.Options.WriteHacks.QuickAck, - ) - - packet, err := m.conn.Read() - if err != nil { - return nil, errors.Annotate(err, "Cannot read packet") - } - - m.logger.Debugw("Read packet length", - "counter", m.readCounter, - "simple_ack", m.req.Options.WriteHacks.SimpleAck, - "quick_ack", m.req.Options.WriteHacks.QuickAck, - "length", len(packet), - ) - - if len(packet) < 4 { - return nil, errors.Annotate(err, "Incorrect packet length") - } - - tag, packet := packet[:4], packet[4:] - switch { - case bytes.Equal(tag, rpc.TagProxyAns): - return m.readProxyAns(packet) - case bytes.Equal(tag, rpc.TagSimpleAck): - return m.readSimpleAck(packet) - case bytes.Equal(tag, rpc.TagCloseExt): - return m.readCloseExt() - } - - return nil, errors.Errorf("Unknown RPC answer %v", tag) -} - -func (m *MTProtoProxy) readProxyAns(data []byte) ([]byte, error) { - if len(data) < 12 { - return nil, errors.Errorf("Incorrect data of proxy answer: %d", len(data)) - } - data = data[12:] - - m.logger.Debugw("Read RPC_PROXY_ANS", - "counter", m.readCounter, - "length", len(data), - ) - - return data, nil -} - -func (m *MTProtoProxy) readSimpleAck(data []byte) ([]byte, error) { - if len(data) != 12 { - return nil, errors.Errorf("Incorrect data of simple ack: %d", len(data)) - } - data = data[8:12] - m.req.Options.WriteHacks.SimpleAck = true - - m.logger.Debugw("Read RPC_SIMPLE_ACK", - "counter", m.readCounter, - "length", len(data), - ) - - return data, nil -} - -func (m *MTProtoProxy) readCloseExt() ([]byte, error) { - m.logger.Debugw("Read RPC_CLOSE_EXT", "counter", m.readCounter) - - return nil, errors.New("Connection has been closed remotely by RPC call") -} - -func (m *MTProtoProxy) Write(p []byte) (int, error) { - defer func() { - m.writeCounter++ - }() - - m.logger.Debugw("Write packet", - "length", len(p), - "counter", m.writeCounter, - "simple_ack", m.req.Options.ReadHacks.SimpleAck, - "quick_ack", m.req.Options.ReadHacks.QuickAck, - ) - - header, flags := m.req.MakeHeader(p) - if ce := m.logger.Desugar().Check(zap.DebugLevel, "RPC_PROXY_REQ header"); ce != nil { - ce.Write( - zap.Int("length", len(p)), - zap.Uint32("counter", m.writeCounter), - zap.Bool("simple_ack", m.req.Options.ReadHacks.QuickAck), - zap.Bool("quick_ack", m.req.Options.ReadHacks.SimpleAck), - zap.String("header", fmt.Sprintf("%v", header.Bytes())), - zap.Stringer("flags", flags), - ) - } - header.Write(p) // nolint: gosec - - if _, err := m.conn.Write(header.Bytes()); err != nil { - return 0, err - } - - return len(p), nil -} - -// Logger returns an instance of the logger for this wrapper. -func (m *MTProtoProxy) Logger() *zap.SugaredLogger { - return m.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (m *MTProtoProxy) LocalAddr() *net.TCPAddr { - return m.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (m *MTProtoProxy) RemoteAddr() *net.TCPAddr { - return m.conn.RemoteAddr() -} - -// Close closes underlying net.Conn instance. -func (m *MTProtoProxy) Close() error { - return m.conn.Close() -} - -// NewMTProtoProxy creates new RPC wrapper. -func NewMTProtoProxy(conn PacketReadWriteCloser, connOpts *mtproto.ConnectionOpts, - adTag []byte) (PacketReadWriteCloser, error) { - req, err := rpc.NewProxyRequest(connOpts.ClientAddr, conn.LocalAddr(), connOpts, adTag) - if err != nil { - return nil, errors.Annotate(err, "Cannot create new RPC proxy request") - } - - return &MTProtoProxy{ - conn: conn, - logger: conn.Logger().Named("mtproto-proxy"), - req: req, - }, nil -} diff --git a/_wrappers/streamcipher.go b/_wrappers/streamcipher.go deleted file mode 100644 index 5c0b137..0000000 --- a/_wrappers/streamcipher.go +++ /dev/null @@ -1,73 +0,0 @@ -package wrappers - -import ( - "bytes" - "crypto/cipher" - "net" - - "github.com/juju/errors" - "go.uber.org/zap" -) - -// StreamCipher is a wrapper which encrypts/decrypts stream with AES-CTR -// (as a part of obfuscated2 protocol). -type StreamCipher struct { - encryptor cipher.Stream - decryptor cipher.Stream - conn StreamReadWriteCloser - logger *zap.SugaredLogger -} - -func (s *StreamCipher) Read(p []byte) (int, error) { - n, err := s.conn.Read(p) - if err != nil { - return 0, errors.Annotate(err, "Cannot read stream ciphered data") - } - s.decryptor.XORKeyStream(p, p[:n]) - - return n, nil -} - -func (s *StreamCipher) Write(p []byte) (int, error) { - buf := streamCipherBufferPool.Get().(*bytes.Buffer) - defer streamCipherBufferPool.Put(buf) - - buf.Reset() - buf.Grow(len(p)) - buf.Write(p) // nolint: gosec - - data := buf.Bytes() - s.encryptor.XORKeyStream(data, data) - - return s.conn.Write(data) -} - -// Logger returns an instance of the logger for this wrapper. -func (s *StreamCipher) Logger() *zap.SugaredLogger { - return s.logger -} - -// LocalAddr returns local address of the underlying net.Conn. -func (s *StreamCipher) LocalAddr() *net.TCPAddr { - return s.conn.LocalAddr() -} - -// RemoteAddr returns remote address of the underlying net.Conn. -func (s *StreamCipher) RemoteAddr() *net.TCPAddr { - return s.conn.RemoteAddr() -} - -// Close closes underlying net.Conn instance. -func (s *StreamCipher) Close() error { - return s.conn.Close() -} - -// NewStreamCipher creates new stream cipher wrapper. -func NewStreamCipher(conn StreamReadWriteCloser, encryptor, decryptor cipher.Stream) StreamReadWriteCloser { - return &StreamCipher{ - conn: conn, - logger: conn.Logger().Named("stream-cipher"), - encryptor: encryptor, - decryptor: decryptor, - } -} diff --git a/_wrappers/streamcipher_pool.go b/_wrappers/streamcipher_pool.go deleted file mode 100644 index 45253c6..0000000 --- a/_wrappers/streamcipher_pool.go +++ /dev/null @@ -1,14 +0,0 @@ -package wrappers - -import ( - "bytes" - "sync" -) - -var ( - streamCipherBufferPool = sync.Pool{ - New: func() interface{} { - return &bytes.Buffer{} - }, - } -) diff --git a/_wrappers/wrap.go b/_wrappers/wrap.go deleted file mode 100644 index 5e9581b..0000000 --- a/_wrappers/wrap.go +++ /dev/null @@ -1,111 +0,0 @@ -package wrappers - -import ( - "io" - "net" - - "go.uber.org/zap" -) - -// Wrap is a base interface for all wrappers in this package. -type Wrap interface { - Logger() *zap.SugaredLogger - LocalAddr() *net.TCPAddr - RemoteAddr() *net.TCPAddr -} - -// Writer is a base interface for writers of this package. -type Writer interface { - io.Writer - Wrap -} - -// Closer is a base interface for wrappers of this package which can -// close connections. -type Closer interface { - io.Closer - Wrap -} - -// WriteCloser is a base interface for wrappers of this package which -// can write to and close connections. -type WriteCloser interface { - io.Closer - Writer -} - -// StreamReader is a base interface for wrappers which can read from the -// stream. -type StreamReader interface { - io.Reader - Wrap -} - -// StreamReadCloser is a base interface for wrappers which can read from -// and close the connections. -type StreamReadCloser interface { - io.Closer - StreamReader -} - -// StreamReadWriter is a base interface for wrappers which can read from -// and write to the connections. -type StreamReadWriter interface { - io.Writer - StreamReader -} - -// StreamWriteCloser is a base interface for wrappers which can write to -// and close the connections. -type StreamWriteCloser interface { - io.WriteCloser - Wrap -} - -// StreamReadWriteCloser is a base interface for stream processors. -type StreamReadWriteCloser interface { - io.Closer - StreamReadWriter -} - -// PacketReader is a base interface for wrappers which reads 'packets'. -// packets are atoms so you either get a packet or you get an error You -// cannot resume reading from packet. -type PacketReader interface { - Read() ([]byte, error) - Wrap -} - -// PacketWriter is a base interface for wrappers which can write packets. -type PacketWriter interface { - io.Writer - Wrap -} - -// PacketReadWriter is a base interface for wrappers which can read from -// and write packets. -type PacketReadWriter interface { - io.Writer - PacketReader -} - -// PacketReadCloser is a base interface for wrappers which can read -// packets and close the connection. -type PacketReadCloser interface { - io.Closer - PacketReader -} - -// PacketWriteCloser is a base interface for wrappers which can write -// packets and close the connection. -type PacketWriteCloser interface { - io.Writer - io.Closer - Wrap -} - -// PacketReadWriteCloser is a base interface for packet processors. -type PacketReadWriteCloser interface { - io.Closer - PacketReadWriter -}