Updates on the policies

Integration with statsd

.gitignore

@@ -8,3 +8,4 @@
 mtg
 vendor/
 version.go
+ccbuilds/

Gopkg.lock

@@ -16,6 +16,12 @@
   packages = ["."]
   revision = "2efee857e7cfd4f3d0138cc3cbb1b4966962b93a"
 
+[[projects]]
+  name = "github.com/beevik/ntp"
+  packages = ["."]
+  revision = "62c80a04de2086884d8296004b6d74ee1846c582"
+  version = "v0.2.0"
+
 [[projects]]
   name = "github.com/davecgh/go-spew"
   packages = ["spew"]
@@ -26,7 +32,7 @@
   branch = "master"
   name = "github.com/dustin/go-humanize"
   packages = ["."]
-  revision = "02af3965c54e8cacf948b97fef38925c4120652c"
+  revision = "9f541cc9db5d55bce703bd99987c9d5cb8eea45e"
 
 [[projects]]
   branch = "master"
@@ -77,15 +83,32 @@
   revision = "eeedf312bc6c57391d84767a4cd413f02a917974"
   version = "v1.8.0"
 
+[[projects]]
+  branch = "master"
+  name = "golang.org/x/net"
+  packages = [
+    "bpf",
+    "internal/iana",
+    "internal/socket",
+    "ipv4"
+  ]
+  revision = "d0887baf81f4598189d4e12a37c6da86f0bba4d0"
+
 [[projects]]
   name = "gopkg.in/alecthomas/kingpin.v2"
   packages = ["."]
   revision = "947dcec5ba9c011838740e680966fd7087a71d0d"
   version = "v2.2.6"
 
+[[projects]]
+  name = "gopkg.in/alexcesaro/statsd.v2"
+  packages = ["."]
+  revision = "7fea3f0d2fab1ad973e641e51dba45443a311a90"
+  version = "v2.0.0"
+
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
-  inputs-digest = "c4fdd3664f683342ad0c2509f4a8bcfe5b267a6e8cdaf36f70d39536bbf89834"
+  inputs-digest = "7fad0f62feb7737b064d85cc4333a1a3e9298faec2afd864b4404f515fc7f17c"
   solver-name = "gps-cdcl"
   solver-version = 1

Gopkg.toml

@@ -48,3 +48,11 @@
 [[constraint]]
   branch = "master"
   name = "github.com/dustin/go-humanize"
+
+[[constraint]]
+  name = "github.com/beevik/ntp"
+  version = "0.2.0"
+
+[[constraint]]
+  name = "gopkg.in/alexcesaro/statsd.v2"
+  version = "2.0.0"

Makefile

@@ -4,7 +4,7 @@ APP_NAME     := $(IMAGE_NAME)
 GOMETALINTER := gometalinter
 
 VENDOR_FILES := $(shell find "$(ROOT_DIR)/vendor" 2>/dev/null || echo -n "vendor")
-CC_BINARIES  := $(shell bash -c "echo -n $(APP_NAME)-{linux,windows,darwin,freebsd,openbsd}-{386,amd64} $(APP_NAME)-linux-{arm,arm64}")
+CC_BINARIES  := $(shell bash -c "echo -n $(APP_NAME)-{linux,freebsd,openbsd}-{386,amd64} $(APP_NAME)-linux-{arm,arm64}")
 APP_DEPS     := version.go $(VENDOR_FILES)
 
 COMMON_BUILD_FLAGS := -ldflags="-s -w"

README.md

@@ -14,7 +14,6 @@ are the most notable:
 * [Official](https://github.com/TelegramMessenger/MTProxy)
 * [Python](https://github.com/alexbers/mtprotoproxy)
 * [Erlang](https://github.com/seriyps/mtproto_proxy)
-* [JS](https://github.com/FreedomPrevails/JSMTProxy)
 
 Almost all of them follow the way how official proxy was build. This
 includes support of multiple secrets, support of promoted channels etc.
@@ -34,7 +33,7 @@ mtg is an implementation in golang which is intended to be:
   software. I also believe that in case of throwout proxies, this feature
   is useless luxury.
 * **Minimum docker image size**
-  Official image is less than 2 megabytes. Literally.
+  Official image is less than 2.5 megabytes. Literally.
 * **No management WebUI**
   This is an implementation of simple lightweight proxy. I won't do that.
 
@@ -51,6 +50,13 @@ To get promoted channel, please contact
 second parameter.
 
 
+# Source code organization
+
+There are 2 main branches:
+
+1. `master` branch contains potentially unstable features
+2. `stable` branch contains stable version. Usually you want to use this branch.
+
 # How to build
 
 ```console
@@ -71,8 +77,22 @@ $ make docker
 
 # Docker image
 
+Docker follows the same policy as the source code organization:
+
+- `latest` mirrors the master branch
+- `stable` mirrors the stable branch
+- tags are for tagged releases
+
 ```console
-$ docker pull nineseconds/mtg
+$ docker pull nineseconds/mtg:latest
+```
+
+```console
+$ docker pull nineseconds/mtg:stable
+```
+
+```console
+$ docker pull nineseconds/mtg:0.10
 ```
 
 # Configuration
@@ -91,18 +111,53 @@ or
 $ head -c 512 /dev/urandom | md5sum | cut -f 1 -d ' '
 ```
 
+## Secure mode
+
+If you want to support new secure mode, please prepend `dd` to the
+secret. For example, secret `cf18fa8ea0267057e2c61a5f7322a8e7` should
+be `ddcf18fa8ea0267057e2c61a5f7322a8e7`. But pay attention that some
+old clients won't support this mode. If this is not your case, I would
+suggest to go with this mode.
+
+Oneliners to generate such secrets:
+
+```console
+$ echo dd$(openssl rand -hex 16)
+```
+
+or
+
+```console
+$ echo dd$(head -c 512 /dev/urandom | md5sum | cut -f 1 -d ' ')
+```
+
+
+# How to run the tool
+
 Now run the tool:
 
 ```console
 $ mtg <secret>
 ```
 
+How to run the tool with ADTag:
+
+```console
+$ mtg <secret> <adtag>
+```
+
 This tool will listen on port 3128 by default with the given secret.
 
 # One-line runner
 
+```console
+$ docker run --name mtg --restart=unless-stopped -p 3128:3128 -p 3129:3129 -d nineseconds/mtg:stable $(openssl rand -hex 16)
 ```
-$ docker run --name mtg --restart=unless-stopped -p 3128:3128 -p 3129:3129 -d nineseconds/mtg $(openssl rand -hex 16)
+
+or in secret mode:
+
+```console
+$ docker run --name mtg --restart=unless-stopped -p 3128:3128 -p 3129:3129 -d nineseconds/mtg:stable dd$(openssl rand -hex 16)
 ```
 
 You will have this tool up and running on port 3128. Now curl
@@ -110,3 +165,33 @@ You will have this tool up and running on port 3128. Now curl
 port 3129 will show you some statistics if you are interested in.
 
 Also, you can use [run-mtg.sh](https://github.com/9seconds/mtg/blob/master/run-mtg.sh) script
+
+
+# statsd integration
+
+mtg provides an integration with statsd, you can enable it with command
+line interface. To enable it, you have to provide IP address of statsd
+service.
+
+Out of the box, mtg supports 2 additional dialects: [InfluxDB](https://www.influxdata.com/blog/getting-started-with-sending-statsd-metrics-to-telegraf-influxdb/)
+and [Datadog](https://docs.datadoghq.com/developers/dogstatsd/).
+
+All metrics are gauges. Here is the list of metrics and their meaning:
+
+| Metric name                     | Unit    | Description                                               |
+|---------------------------------|---------|-----------------------------------------------------------|
+| `connections.abridged.ipv4`     | number  | The number of active abridged IPv4 connections            |
+| `connections.abridged.ipv6`     | number  | The number of active abridged IPv6 connections            |
+| `connections.intermediate.ipv4` | number  | The number of active intermediate IPv4 connections        |
+| `connections.intermediate.ipv6` | number  | The number of active intermediate IPv6 connections        |
+| `connections.secure.ipv4`       | number  | The number of active secure intermediate IPv4 connections |
+| `connections.secure.ipv6`       | number  | The number of active secure intermediate IPv6 connections |
+| `crashes`                       | number  | An amount of crashes in client handlers                   |
+| `traffic.ingress`               | bytes   | Ingress traffic from the start of application (incoming)  |
+| `traffic.egress`                | bytes   | Egress traffic from the start of application (outgoing)   |
+| `speed.ingress`                 | bytes/s | Ingress bandwidth of the latest second (incoming traffic) |
+| `speed.egress`                  | bytes/s | Egress bandwidth of the latest second (outgoing traffic)  |
+
+All metrics are prefixed with given prefix. Default prefix is `mtg`.
+With such prefix metric name `traffic.ingress`, for example, would be
+`mtg.traffic.ingress`.

client/middle.go

@@ -17,9 +17,16 @@ func MiddleInit(socket net.Conn, connID string, conf *config.Config) (wrappers.W
 	}
 	connStream := conn.(wrappers.StreamReadWriteCloser)
 
-	newConn := wrappers.NewMTProtoAbridged(connStream, opts)
+	var newConn wrappers.PacketReadWriteCloser
-	if opts.ConnectionType != mtproto.ConnectionTypeAbridged {
+	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

config/config.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 	"net"
 	"strconv"
+	"strings"
 
 	"github.com/juju/errors"
+	statsd "gopkg.in/alexcesaro/statsd.v2"
 )
 
 // Buffer sizes define internal socket buffer sizes.
@@ -19,6 +21,7 @@ const (
 type Config struct {
 	Debug      bool
 	Verbose    bool
+	SecureMode bool
 
 	BindPort       uint16
 	PublicIPv4Port uint16
@@ -30,6 +33,14 @@ type Config struct {
 	PublicIPv6 net.IP
 	StatsIP    net.IP
 
+	StatsD struct {
+		Addr       net.Addr
+		Prefix     string
+		Tags       map[string]string
+		TagsFormat statsd.TagFormat
+		Enabled    bool
+	}
+
 	Secret []byte
 	AdTag  []byte
 }
@@ -46,6 +57,7 @@ type URLs struct {
 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.
@@ -64,15 +76,32 @@ 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, c.Secret)
+		urls.IPv4 = getURLs(c.PublicIPv4, c.PublicIPv4Port, secret)
 	}
 	if c.PublicIPv6 != nil {
-		urls.IPv6 = getURLs(c.PublicIPv6, c.PublicIPv6Port, c.Secret)
+		urls.IPv6 = getURLs(c.PublicIPv6, c.PublicIPv6Port, secret)
 	}
+	urls.BotSecret = c.BotSecretString()
 
 	return urls
 }
@@ -89,8 +118,14 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo
 	publicIPv4 net.IP, PublicIPv4Port uint16,
 	publicIPv6 net.IP, publicIPv6Port uint16,
 	statsIP net.IP, statsPort uint16,
-	secret, adtag string) (*Config, error) {
+	secret, adtag string,
-	if len(secret) != 32 {
+	statsdIP string, statsdPort uint16, statsdNetwork string, statsdPrefix string,
+	statsdTagsFormat string, statsdTags map[string]string) (*Config, error) {
+	secureMode := false
+	if strings.HasPrefix(secret, "dd") && len(secret) == 34 {
+		secureMode = true
+		secret = strings.TrimPrefix(secret, "dd")
+	} else if len(secret) != 32 {
 		return nil, errors.New("Telegram demands secret of length 32")
 	}
 	secretBytes, err := hex.DecodeString(secret)
@@ -147,6 +182,38 @@ func NewConfig(debug, verbose bool, // nolint: gocyclo
 		StatsPort:      statsPort,
 		Secret:         secretBytes,
 		AdTag:          adTagBytes,
+		SecureMode:     secureMode,
+	}
+
+	if statsdIP != "" {
+		conf.StatsD.Enabled = true
+		conf.StatsD.Prefix = statsdPrefix
+		conf.StatsD.Tags = statsdTags
+
+		var addr net.Addr
+		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

config/urls.go

@@ -1,17 +1,16 @@
 package config
 
 import (
-	"encoding/hex"
 	"net"
 	"net/url"
 	"strconv"
 )
 
-func getURLs(addr net.IP, port uint16, secret []byte) (urls URLs) {
+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", hex.EncodeToString(secret))
+	values.Set("secret", secret)
 
 	urls.TG = makeTGURL(values)
 	urls.TMe = makeTMeURL(values)

main.go

@@ -4,20 +4,22 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"math/rand"
 	"os"
 	"syscall"
 	"time"
 
+	"github.com/juju/errors"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
 
 	"github.com/9seconds/mtg/config"
+	"github.com/9seconds/mtg/ntp"
 	"github.com/9seconds/mtg/proxy"
 	"github.com/9seconds/mtg/stats"
-	"github.com/juju/errors"
 )
 
 var (
@@ -59,7 +61,7 @@ var (
 			Envar("MTG_IPV6_PORT").
 			Uint16()
 
-	statsIP = app.Flag("stats-ip", "Which IP bind stats server to").
+	statsIP = app.Flag("stats-ip", "Which IP bind stats server to.").
 		Short('t').
 		Envar("MTG_STATS_IP").
 		Default("127.0.0.1").
@@ -70,6 +72,28 @@ var (
 			Default("3129").
 			Uint16()
 
+	statsdIP = app.Flag("statsd-ip", "Which IP should we use for working with statsd.").
+			Envar("MTG_STATSD_IP").
+			String()
+	statsdPort = app.Flag("statsd-port", "Which port should we use for working with statsd.").
+			Envar("MTG_STATSD_PORT").
+			Default("8125").
+			Uint16()
+	statsdNetwork = app.Flag("statsd-network", "Which network is used to work with statsd. Only 'tcp' and 'udp' are supported.").
+			Envar("MTG_STATSD_NETWORK").
+			Default("udp").
+			String()
+	statsdPrefix = app.Flag("statsd-prefix", "Which bucket prefix should we use for sending stats to statsd.").
+			Envar("MTG_STATSD_PREFIX").
+			Default("mtg").
+			String()
+	statsdTagsFormat = app.Flag("statsd-tags-format", "Which tag format should we use to send stats metrics. Valid options are 'datadog' and 'influxdb'.").
+				Envar("MTG_STATSD_TAGS_FORMAT").
+				String()
+	statsdTags = app.Flag("statsd-tags", "Tags to use for working with statsd (specified as 'key=value').").
+			Envar("MTG_STATSD_TAGS").
+			StringMap()
+
 	secret = app.Arg("secret", "Secret of this proxy.").Required().String()
 	adtag  = app.Arg("adtag", "ADTag of the proxy.").String()
 )
@@ -94,6 +118,8 @@ func main() {
 		*publicIPv6, *publicIPv6Port,
 		*statsIP, *statsPort,
 		*secret, *adtag,
+		*statsdIP, *statsdPort, *statsdNetwork, *statsdPrefix,
+		*statsdTagsFormat, *statsdTags,
 	)
 	if err != nil {
 		usage(err.Error())
@@ -120,11 +146,21 @@ func main() {
 
 	if conf.UseMiddleProxy() {
 		zap.S().Infow("Use middle proxy connection to Telegram")
+		if diff, err := ntp.Fetch(); err != nil {
+			zap.S().Warnw("Could not fetch time data from NTP")
+		} else {
+			if diff >= time.Second {
+				usage(fmt.Sprintf("You choose to use middle proxy but your clock drift (%s) is bigger than 1 second. Please, sync your time", diff))
+			}
+			go ntp.AutoUpdate()
+		}
 	} else {
 		zap.S().Infow("Use direct connection to Telegram")
 	}
 
-	go stats.Start(conf)
+	if err := stats.Start(conf); err != nil {
+		panic(err)
+	}
 
 	server := proxy.NewProxy(conf)
 	if err := server.Serve(); err != nil {

mtproto/connection_options.go

@@ -39,6 +39,7 @@ const (
 	ConnectionTypeUnknown ConnectionType = iota
 	ConnectionTypeAbridged
 	ConnectionTypeIntermediate
+	ConnectionTypeSecure
 )
 
 // ConnectionProtocol* define which connection protocols to use.
@@ -53,6 +54,7 @@ const (
 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.
@@ -62,6 +64,8 @@ func (t ConnectionType) Tag() ([]byte, error) {
 		return ConnectionTagAbridged, nil
 	case ConnectionTypeIntermediate:
 		return ConnectionTagIntermediate, nil
+	case ConnectionTypeSecure:
+		return ConnectionTagSecure, nil
 	default:
 		return nil, errors.Errorf("Unknown connection type %d", t)
 	}
@@ -75,6 +79,9 @@ func ConnectionTagFromHandshake(magic []byte) (ConnectionType, error) {
 	if bytes.Equal(magic, ConnectionTagAbridged) {
 		return ConnectionTypeAbridged, nil
 	}
+	if bytes.Equal(magic, ConnectionTagSecure) {
+		return ConnectionTypeSecure, nil
+	}
 
 	return ConnectionTypeUnknown, errors.New("Unknown handshake protocol")
 }

mtproto/rpc/proxy_flags.go

@@ -15,6 +15,7 @@ const (
 	proxyRequestFlagsIntermediate                   = 0x20000000
 	proxyRequestFlagsAbdridged                      = 0x40000000
 	proxyRequestFlagsQuickAck                       = 0x80000000
+	proxyRequestFlagsPad                            = 0x8000000
 )
 
 var proxyRequestFlagsEncryptedPrefix [8]byte
@@ -50,6 +51,9 @@ func (r proxyRequestFlags) String() string {
 	if r&proxyRequestFlagsQuickAck != 0 {
 		flags = append(flags, "QUICK_ACK")
 	}
+	if r&proxyRequestFlagsPad != 0 {
+		flags = append(flags, "PAD")
+	}
 
 	return strings.Join(flags, " | ")
 }

mtproto/rpc/proxy_request.go

@@ -72,6 +72,10 @@ func NewProxyRequest(clientAddr, ownAddr *net.TCPAddr, opts *mtproto.ConnectionO
 		flags |= proxyRequestFlagsAbdridged
 	case mtproto.ConnectionTypeIntermediate:
 		flags |= proxyRequestFlagsIntermediate
+	case mtproto.ConnectionTypeSecure:
+		flags |= proxyRequestFlagsIntermediate | proxyRequestFlagsPad
+	default:
+		panic("Unknown connection type")
 	}
 
 	request := &ProxyRequest{

ntp/ntp.go

@@ -0,0 +1,60 @@
+package ntp
+
+import (
+	"math/rand"
+	"time"
+
+	"github.com/beevik/ntp"
+	"github.com/juju/errors"
+	"go.uber.org/zap"
+)
+
+const autoUpdatePeriod = time.Minute
+
+var ntpEndpoints = []string{
+	"0.pool.ntp.org",
+	"1.pool.ntp.org",
+	"2.pool.ntp.org",
+	"3.pool.ntp.org",
+}
+
+// Fetch fetches the data on time drift.
+func Fetch() (time.Duration, error) {
+	url := ntpEndpoints[rand.Intn(len(ntpEndpoints))]
+	resp, err := ntp.Query(url)
+	if err != nil {
+		return 0, errors.Annotatef(err, "Cannot fetch NTP server %s", url)
+	}
+
+	offsetInt := int64(resp.ClockOffset)
+	if offsetInt < 0 {
+		offsetInt = -offsetInt
+	}
+	offset := time.Duration(offsetInt)
+
+	return offset, nil
+}
+
+// AutoUpdate runs periodic check of current time .drift state.
+func AutoUpdate() {
+	logger := zap.S().Named("ntp")
+
+	for range time.Tick(autoUpdatePeriod) {
+		diff, err := Fetch()
+		if err != nil {
+			logger.Debugw("Cannot fetch time from NTP", "error", err)
+			continue
+		}
+
+		switch {
+		case diff < 400*time.Millisecond:
+			logger.Debugw("NTP time drift", "value", diff.String())
+		case diff < 600*time.Millisecond:
+			logger.Infow("NTP time drift", "value", diff.String())
+		case diff < 800*time.Millisecond:
+			logger.Warnw("NTP time drift", "value", diff.String())
+		default:
+			logger.Errorw("NTP time drift", "value", diff.String())
+		}
+	}
+}

stats/channels.go

@@ -40,7 +40,7 @@ func crashManager() {
 	}
 }
 
-func connectionManager() { // nolint: gocyclo
+func connectionManager() {
 	for event := range connectionsChan {
 		instance.mutex.RLock()
 
@@ -53,27 +53,21 @@ func connectionManager() { // nolint: gocyclo
 		switch event.connectionType {
 		case mtproto.ConnectionTypeAbridged:
 			if isIPv4 {
-				instance.ActiveConnections.Abridged.IPv4 += inc
+				instance.Connections.Abridged.IPv4 += inc
-				if event.connected {
-					instance.AllConnections.Abridged.IPv4 += inc
-				}
 			} else {
-				instance.ActiveConnections.Abridged.IPv6 += inc
+				instance.Connections.Abridged.IPv6 += inc
-				if event.connected {
-					instance.AllConnections.Abridged.IPv6 += inc
 			}
+		case mtproto.ConnectionTypeSecure:
+			if isIPv4 {
+				instance.Connections.Secure.IPv4 += inc
+			} else {
+				instance.Connections.Secure.IPv6 += inc
 			}
 		default:
 			if isIPv4 {
-				instance.ActiveConnections.Intermediate.IPv4 += inc
+				instance.Connections.Intermediate.IPv4 += inc
-				if event.connected {
-					instance.AllConnections.Intermediate.IPv4 += inc
-				}
 			} else {
-				instance.ActiveConnections.Intermediate.IPv6 += inc
+				instance.Connections.Intermediate.IPv6 += inc
-				if event.connected {
-					instance.AllConnections.Intermediate.IPv6 += inc
-				}
 			}
 		}
 

stats/server.go

@@ -14,7 +14,7 @@ import (
 var instance *stats
 
 // Start starts new statisitcs server.
-func Start(conf *config.Config) {
+func Start(conf *config.Config) error {
 	log := zap.S().Named("stats")
 
 	instance = &stats{
@@ -23,6 +23,14 @@ func Start(conf *config.Config) {
 		mutex:  &sync.RWMutex{},
 	}
 
+	if conf.StatsD.Enabled {
+		client, err := newStatsd(conf)
+		if err != nil {
+			return err
+		}
+		go client.run()
+	}
+
 	go crashManager()
 	go connectionManager()
 	go trafficManager()
@@ -51,7 +59,11 @@ func Start(conf *config.Config) {
 		}
 	})
 
+	go func() {
 		if err := http.ListenAndServe(conf.StatAddr(), nil); err != nil {
 			log.Fatalw("Stats server has been stopped", "error", err)
 		}
+	}()
+
+	return nil
 }

stats/stats.go

@@ -52,20 +52,23 @@ type connections struct {
 	All          connectionType `json:"all"`
 	Abridged     connectionType `json:"abridged"`
 	Intermediate connectionType `json:"intermediate"`
+	Secure       connectionType `json:"secure"`
 }
 
 func (c connections) MarshalJSON() ([]byte, error) {
-	c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4
+	c.All.IPv4 = c.Abridged.IPv4 + c.Intermediate.IPv4 + c.Secure.IPv4
-	c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6
+	c.All.IPv6 = c.Abridged.IPv6 + c.Intermediate.IPv6 + c.Secure.IPv6
 
 	value := struct {
 		All          connectionType `json:"all"`
 		Abridged     connectionType `json:"abridged"`
 		Intermediate connectionType `json:"intermediate"`
+		Secure       connectionType `json:"secure"`
 	}{
 		All:          c.All,
 		Abridged:     c.Abridged,
 		Intermediate: c.Intermediate,
+		Secure:       c.Secure,
 	}
 
 	return json.Marshal(value)
@@ -88,8 +91,7 @@ type speed struct {
 
 type stats struct {
 	URLs        config.IPURLs `json:"urls"`
-	ActiveConnections connections   `json:"active_connections"`
+	Connections connections   `json:"connections"`
-	AllConnections    connections   `json:"all_connections"`
 	Traffic     traffic       `json:"traffic"`
 	Speed       speed         `json:"speed"`
 	Uptime      uptime        `json:"uptime"`

stats/statsd.go

@@ -0,0 +1,79 @@
+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.mutex.Lock()
+
+		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, uint64(instance.Traffic.Ingress))
+		s.client.Gauge(statsdTrafficEgress, uint64(instance.Traffic.Egress))
+		s.client.Gauge(statsdSpeedIngress, uint64(instance.Speed.Ingress))
+		s.client.Gauge(statsdSpeedEgress, uint64(instance.Speed.Egress))
+		s.client.Gauge(statsdCrashes, instance.Crashes)
+
+		instance.mutex.Unlock()
+	}
+}
+
+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
+}

wrappers/mtproto_intermediate.go

@@ -62,10 +62,6 @@ func (m *MTProtoIntermediate) Read() ([]byte, error) {
 		return nil, errors.Annotate(err, "Cannot read the message")
 	}
 
-	if length%4 != 0 {
-		length -= length % 4
-	}
-
 	return buf.Bytes()[:length], nil
 }
 
@@ -80,7 +76,7 @@ func (m *MTProtoIntermediate) Write(p []byte) (int, error) {
 		"counter", m.writeCounter,
 	)
 
-	if m.opts.ReadHacks.SimpleAck {
+	if m.opts.WriteHacks.SimpleAck {
 		return m.conn.Write(p)
 	}
 

wrappers/mtproto_intermediate_secure.go

@@ -0,0 +1,74 @@
+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
+	buf.Write(p)
+	buf.Write(make([]byte, paddingLength))
+
+	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,
+		},
+	}
+}