Merge pull request #122 from 9seconds/close-connections

Make cloaking more robust

faketls/client_protocol.go

 	"io"
 	"net"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/9seconds/mtg/antireplay"
 }
 
 func (c *ClientProtocol) cloakHost(clientConn io.ReadWriteCloser) {
+	stats.Stats.CloakedRequest()
+
 	addr := net.JoinHostPort(config.C.CloakHost, strconv.Itoa(config.C.CloakPort))
 	hostConn, err := net.Dial("tcp", addr)
 
 		return
 	}
 
-	defer hostConn.Close()
-
-	wg := &sync.WaitGroup{}
-	wg.Add(2)
-
-	go c.pipe(hostConn, clientConn, wg)
-
-	go c.pipe(clientConn, hostConn, wg)
-
-	wg.Wait()
-}
-
-func (c *ClientProtocol) pipe(dst io.WriteCloser, src io.Reader, wg *sync.WaitGroup) {
-	defer func() {
-		wg.Done()
-		dst.Close()
-	}()
-
-	io.Copy(dst, src) // nolint: errcheck
+	cloak(clientConn, hostConn)
 }
 
 func MakeClientProtocol() protocol.ClientProtocol {

faketls/cloak.go

+package faketls
+
+import (
+	"context"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/9seconds/mtg/wrappers/rwc"
+)
+
+const (
+	cloakLastActivityTimeout = 5 * time.Second
+	cloakMaxTimeout          = 30 * time.Second
+)
+
+func cloak(one, another io.ReadWriteCloser) {
+	defer func() {
+		one.Close()
+		another.Close()
+	}()
+
+	channelPing := make(chan struct{}, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	one = rwc.NewPing(ctx, one, channelPing)
+	another = rwc.NewPing(ctx, another, channelPing)
+	wg := &sync.WaitGroup{}
+
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		io.Copy(one, another) // nolint: errcheck
+	}()
+
+	go func() {
+		defer wg.Done()
+		io.Copy(another, one) // nolint: errcheck
+	}()
+
+	go func() {
+		wg.Wait()
+		cancel()
+	}()
+
+	go func() {
+		lastActivityTimer := time.NewTimer(cloakLastActivityTimeout)
+		defer lastActivityTimer.Stop()
+
+		maxTimer := time.NewTimer(cloakMaxTimeout)
+		defer maxTimer.Stop()
+
+		for {
+			select {
+			case <-channelPing:
+				lastActivityTimer.Stop()
+				lastActivityTimer = time.NewTimer(cloakLastActivityTimeout)
+			case <-ctx.Done():
+				return
+			case <-lastActivityTimer.C:
+				cancel()
+				return
+			case <-maxTimer.C:
+				cancel()
+				return
+			}
+		}
+	}()
+
+	<-ctx.Done()
+}

go.mod

 go 1.13
 
 require (
-	github.com/VictoriaMetrics/fastcache v1.5.2
+	github.com/VictoriaMetrics/fastcache v1.5.4
 	github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d
 	github.com/beevik/ntp v0.2.0
-	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/prometheus/client_golang v1.2.1
-	github.com/prometheus/procfs v0.0.7 // indirect
+	github.com/prometheus/client_model v0.0.0-20191202183732-d1d2010b5bee // indirect
+	github.com/prometheus/procfs v0.0.8 // indirect
 	github.com/smira/go-statsd v1.3.1
 	go.uber.org/atomic v1.5.1 // indirect
 	go.uber.org/multierr v1.4.0 // indirect
 	go.uber.org/zap v1.13.0
-	golang.org/x/crypto v0.0.0-20191122220453-ac88ee75c92c
-	golang.org/x/net v0.0.0-20191124235446-72fef5d5e266 // indirect
-	golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e
-	golang.org/x/tools v0.0.0-20191125011157-cc15fab314e3 // indirect
+	golang.org/x/crypto v0.0.0-20191202143827-86a70503ff7e
+	golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f // indirect
+	golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933 // indirect
+	golang.org/x/sys v0.0.0-20191128015809-6d18c012aee9
+	golang.org/x/tools v0.0.0-20191203051722-db047d72ee39 // indirect
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 )

go.sum

 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
-github.com/OneOfOne/xxhash v1.2.5/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
-github.com/VictoriaMetrics/fastcache v1.5.2 h1:Erd8iIuBAL9kke8JzM4+WxkKuFkHh3ktwLanJvDgR44=
-github.com/VictoriaMetrics/fastcache v1.5.2/go.mod h1:+jv9Ckb+za/P1ZRg/sulP5Ni1v49daAVERr0H3CuscE=
+github.com/VictoriaMetrics/fastcache v1.5.4 h1:0BaXbRH01RycJk79OOBwMCXlNryko9z4yEf6RqbP+Xo=
+github.com/VictoriaMetrics/fastcache v1.5.4/go.mod h1:ptDBkNMQI4RtmVo8VS/XwRY6RoTu1dAWCbrk+6WsEM8=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
-github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
-github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
-github.com/cespare/xxhash/v2 v2.0.1-0.20190104013014-3767db7a7e18/go.mod h1:HD5P3vAIAh+Y2GAxg0PrPN1P8WkepXGpjbUPDHJqqKM=
 github.com/cespare/xxhash/v2 v2.1.0 h1:yTUvW7Vhb89inJ+8irsUqiWjh8iT6sQPZiQzI6ReGkA=
 github.com/cespare/xxhash/v2 v2.1.0/go.mod h1:dgIUBU3pDso/gPgZ1osOZ0iQf77oPR28Tjxl5dIMyVM=
 github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 h1:gQz4mCbXsO+nc9n1hCxHcGA3Zx3Eo+UHZoInFGUIXNM=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.0.0-20191202183732-d1d2010b5bee h1:iBZPTYkGLvdu6+A5TsMUJQkQX9Ad4aCEnSQtdxPuTCQ=
+github.com/prometheus/client_model v0.0.0-20191202183732-d1d2010b5bee/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.4.1 h1:K0MGApIoQvMw27RTdJkPbr3JZ7DNbtxQNyi5STVM6Kw=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.7.0 h1:L+1lyG48J1zAQXA3RBX/nG/B3gjlHq0zTt2tlbJLyCY=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.5 h1:3+auTFlqw+ZaQYJARz6ArODtkaIwtvBTx3N2NehQlL8=
 github.com/prometheus/procfs v0.0.5/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
-github.com/prometheus/procfs v0.0.7 h1:RS5GAlMbnkWkhs4+bPocMTmGjYkuCY5djjqEDdXOhcQ=
-github.com/prometheus/procfs v0.0.7/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
+github.com/prometheus/procfs v0.0.8 h1:+fpWZdT24pJBiqJdAwYBjPSk+5YmQzYNPYzQsdzLkt8=
+github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/smira/go-statsd v1.3.1 h1:JalGiHNdK7GqVAPpg7j0Kwp2jZrz/fCg/B4ZuNuBY2w=
 github.com/smira/go-statsd v1.3.1/go.mod h1:1srXJ9/pbnN04G8f4F1jUzsGOnwkPKXciyqpewGlkC4=
-github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
-github.com/spaolacci/murmur3 v1.0.1-0.20190317074736-539464a789e9/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20191122220453-ac88ee75c92c h1:/nJuwDLoL/zrqY6gf57vxC+Pi+pZ8bfhpPkicO5H7W4=
-golang.org/x/crypto v0.0.0-20191122220453-ac88ee75c92c/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20191202143827-86a70503ff7e h1:egKlR8l7Nu9vHGWbcUV8lqR4987UfUbBd7GbhqGzNYU=
+golang.org/x/crypto v0.0.0-20191202143827-86a70503ff7e/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f h1:J5lckAjkw6qYlOZNj90mLYNTEKDvWeuc1yieZ8qUzUE=
+golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f/go.mod h1:5qLYkcX4OjUUV8bRuDixDT3tpyyb+LUpUlRWLxfhWrs=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20191124235446-72fef5d5e266 h1:QuOiA7GCO0OSDzlNlFyOWOywDsjuzW8M2yvBfCqw+cY=
-golang.org/x/net v0.0.0-20191124235446-72fef5d5e266/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933 h1:e6HwijUxhDe+hPNjZQQn9bA5PW3vNmnN64U2ZW759Lk=
+golang.org/x/net v0.0.0-20191126235420-ef20fe5d7933/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
 golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e h1:N7DeIrjYszNmSW409R3frPPwglRwMkXSBzwVbkOjLLA=
-golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191128015809-6d18c012aee9 h1:ZBzSG/7F4eNKz2L3GE9o300RX0Az1Bw5HF7PDraD+qU=
+golang.org/x/sys v0.0.0-20191128015809-6d18c012aee9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20191125011157-cc15fab314e3 h1:aHkNOJLg6a84bdLJN1yjqMSTadeAuaudhEPNSkLAWoA=
-golang.org/x/tools v0.0.0-20191125011157-cc15fab314e3/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191125144606-a911d9008d1f/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20191203051722-db047d72ee39 h1:zARK4PTmTfx1BC6iKP21qIRjz0nFzFj4ZAlbUy6Q6pM=
+golang.org/x/tools v0.0.0-20191203051722-db047d72ee39/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=

proxy/proxy.go

 	clientConn, err := clientProtocol.Handshake(clientConn)
 
 	if err != nil {
+		stats.Stats.AuthenticationFailed()
 		logger.Warnw("Cannot perform client handshake", "error", err)
+
 		return
 	}
 

stats/interfaces.go

 	ReplayDetected()
 }
 
+type AuthenticationFailedInterface interface {
+	AuthenticationFailed()
+}
+
+type CloakedRequestInterface interface {
+	CloakedRequest()
+}
+
 type Interface interface {
 	IngressTrafficInterface
 	EgressTrafficInterface
 	TelegramDisconnectedInterface
 	CrashInterface
 	ReplayDetectedInterface
+	AuthenticationFailedInterface
+	CloakedRequestInterface
 }

stats/multi_stats.go

 		go m[i].ReplayDetected()
 	}
 }
+
+func (m multiStats) AuthenticationFailed() {
+	for i := range m {
+		go m[i].AuthenticationFailed()
+	}
+}
+
+func (m multiStats) CloakedRequest() {
+	for i := range m {
+		go m[i].CloakedRequest()
+	}
+}

stats/stats_prometheus.go

 )
 
 type statsPrometheus struct {
-	connections         *prometheus.GaugeVec
-	telegramConnections *prometheus.GaugeVec
-	traffic             *prometheus.GaugeVec
-	crashes             prometheus.Counter
-	replayAttacks       prometheus.Counter
+	connections          *prometheus.GaugeVec
+	telegramConnections  *prometheus.GaugeVec
+	traffic              *prometheus.GaugeVec
+	crashes              prometheus.Counter
+	replayAttacks        prometheus.Counter
+	authenticationFailed prometheus.Counter
+	cloakedRequests      prometheus.Counter
 }
 
 func (s *statsPrometheus) IngressTraffic(traffic int) {
 	s.replayAttacks.Inc()
 }
 
+func (s *statsPrometheus) AuthenticationFailed() {
+	s.authenticationFailed.Inc()
+}
+
+func (s *statsPrometheus) CloakedRequest() {
+	s.cloakedRequests.Inc()
+}
+
 func newStatsPrometheus(mux *http.ServeMux) Interface {
 	registry := prometheus.NewPedanticRegistry()
 
 			Name:      "replay_attacks",
 			Help:      "How many replay attacks were prevented.",
 		}),
+		authenticationFailed: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: config.C.StatsNamespace,
+			Name:      "authentication_failed",
+			Help:      "How many authentication failed events we've seen.",
+		}),
+		cloakedRequests: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: config.C.StatsNamespace,
+			Name:      "cloaked_requests",
+			Help:      "How many requests were proxified during cloaking.",
+		}),
 	}
 
 	registry.MustRegister(instance.connections)
 	registry.MustRegister(instance.traffic)
 	registry.MustRegister(instance.crashes)
 	registry.MustRegister(instance.replayAttacks)
+	registry.MustRegister(instance.authenticationFailed)
+	registry.MustRegister(instance.cloakedRequests)
 
 	handler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{})
 	mux.Handle("/", handler)

stats/stats_statsd.go

 	s.gauge("replay_attacks", 1)
 }
 
+func (s *statsStatsd) AuthenticationFailed() {
+	s.gauge("authentication_failed", 1)
+}
+
+func (s *statsStatsd) CloakedRequest() {
+	s.gauge("cloaked_requests", 1)
+}
+
 func (s *statsStatsd) gauge(metric string, value int64, tags ...*statsStatsdTag) {
 	key, tagList := s.prepareVals(metric, tags)
 	s.initGauge(metric, key, tagList)

wrappers/rwc/ping.go

+package rwc
+
+import (
+	"context"
+	"io"
+)
+
+type wrapperPing struct {
+	parent      io.ReadWriteCloser
+	ctx         context.Context
+	channelPing chan<- struct{}
+}
+
+func (w *wrapperPing) Read(p []byte) (int, error) {
+	n, err := w.parent.Read(p)
+	if err == nil {
+		select {
+		case <-w.ctx.Done():
+		case w.channelPing <- struct{}{}:
+		}
+	}
+
+	return n, err
+}
+
+func (w *wrapperPing) Write(p []byte) (int, error) {
+	n, err := w.parent.Write(p)
+	if err == nil {
+		select {
+		case <-w.ctx.Done():
+		case w.channelPing <- struct{}{}:
+		}
+	}
+
+	return n, err
+}
+
+func (w *wrapperPing) Close() error {
+	return w.parent.Close()
+}
+
+func NewPing(ctx context.Context, parent io.ReadWriteCloser, channelPing chan<- struct{}) io.ReadWriteCloser {
+	return &wrapperPing{
+		parent:      parent,
+		ctx:         ctx,
+		channelPing: channelPing,
+	}
+}