Make SecretConnection thread safe (#3111)

* p2p/conn: add failing tests

* p2p/conn: make SecretConnection thread safe

* changelog

* fix from review

CHANGELOG_PENDING.md

@@ -1,4 +1,4 @@
-## v0.27.4
+## v0.28.0
 
 *TBD*
 
@@ -15,9 +15,9 @@ Special thanks to external contributors on this release:
 
 * Apps
 
-* Go API  
-- [types] \#2926 memoize consensus public key on initialization of remote signer and return the memoized key on 
-`PrivValidator.GetPubKey()` instead of requesting it again 
+* Go API
+- [types] \#2926 memoize consensus public key on initialization of remote signer and return the memoized key on
+`PrivValidator.GetPubKey()` instead of requesting it again
 - [types] \#2981 Remove `PrivValidator.GetAddress()`
 
 * Blockchain Protocol
@@ -29,6 +29,7 @@ Special thanks to external contributors on this release:
 - [privval] \#1181 Split immutable and mutable parts of priv_validator.json
 
 ### IMPROVEMENTS:
+- [p2p/conn] \#3111 make SecretConnection thread safe
 - [rpc] \#3047 Include peer's remote IP in `/net_info`
 
 ### BUG FIXES:

p2p/conn/secret_connection.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"io"
 	"net"
+	"sync"
 	"time"
 
 	"golang.org/x/crypto/chacha20poly1305"
@@ -27,20 +28,36 @@ const aeadSizeOverhead = 16 // overhead of poly 1305 authentication tag
 const aeadKeySize = chacha20poly1305.KeySize
 const aeadNonceSize = chacha20poly1305.NonceSize
 
-// SecretConnection implements net.conn.
+// SecretConnection implements net.Conn.
 // It is an implementation of the STS protocol.
-// Note we do not (yet) assume that a remote peer's pubkey
-// is known ahead of time, and thus we are technically
-// still vulnerable to MITM. (TODO!)
-// See docs/sts-final.pdf for more info
+// See https://github.com/tendermint/tendermint/blob/0.1/docs/sts-final.pdf for
+// details on the protocol.
+//
+// Consumers of the SecretConnection are responsible for authenticating
+// the remote peer's pubkey against known information, like a nodeID.
+// Otherwise they are vulnerable to MITM.
+// (TODO(ismail): see also https://github.com/tendermint/tendermint/issues/3010)
 type SecretConnection struct {
-	conn       io.ReadWriteCloser
-	recvBuffer []byte
-	recvNonce  *[aeadNonceSize]byte
-	sendNonce  *[aeadNonceSize]byte
+
+	// immutable
 	recvSecret *[aeadKeySize]byte
 	sendSecret *[aeadKeySize]byte
 	remPubKey  crypto.PubKey
+	conn       io.ReadWriteCloser
+
+	// net.Conn must be thread safe:
+	// https://golang.org/pkg/net/#Conn.
+	// Since we have internal mutable state,
+	// we need mtxs. But recv and send states
+	// are independent, so we can use two mtxs.
+	// All .Read are covered by recvMtx,
+	// all .Write are covered by sendMtx.
+	recvMtx    sync.Mutex
+	recvBuffer []byte
+	recvNonce  *[aeadNonceSize]byte
+
+	sendMtx   sync.Mutex
+	sendNonce *[aeadNonceSize]byte
 }
 
 // MakeSecretConnection performs handshake and returns a new authenticated
@@ -109,9 +126,12 @@ func (sc *SecretConnection) RemotePubKey() crypto.PubKey {
 	return sc.remPubKey
 }
 
-// Writes encrypted frames of `sealedFrameSize`
-// CONTRACT: data smaller than dataMaxSize is read atomically.
+// Writes encrypted frames of `totalFrameSize + aeadSizeOverhead`.
+// CONTRACT: data smaller than dataMaxSize is written atomically.
 func (sc *SecretConnection) Write(data []byte) (n int, err error) {
+	sc.sendMtx.Lock()
+	defer sc.sendMtx.Unlock()
+
 	for 0 < len(data) {
 		var frame = make([]byte, totalFrameSize)
 		var chunk []byte
@@ -130,6 +150,7 @@ func (sc *SecretConnection) Write(data []byte) (n int, err error) {
 		if err != nil {
 			return n, errors.New("Invalid SecretConnection Key")
 		}
+
 		// encrypt the frame
 		var sealedFrame = make([]byte, aeadSizeOverhead+totalFrameSize)
 		aead.Seal(sealedFrame[:0], sc.sendNonce[:], frame, nil)
@@ -147,23 +168,30 @@ func (sc *SecretConnection) Write(data []byte) (n int, err error) {
 
 // CONTRACT: data smaller than dataMaxSize is read atomically.
 func (sc *SecretConnection) Read(data []byte) (n int, err error) {
+	sc.recvMtx.Lock()
+	defer sc.recvMtx.Unlock()
+
+	// read off and update the recvBuffer, if non-empty
 	if 0 < len(sc.recvBuffer) {
 		n = copy(data, sc.recvBuffer)
 		sc.recvBuffer = sc.recvBuffer[n:]
 		return
 	}
 
-	aead, err := chacha20poly1305.New(sc.recvSecret[:])
-	if err != nil {
-		return n, errors.New("Invalid SecretConnection Key")
-	}
+	// read off the conn
 	sealedFrame := make([]byte, totalFrameSize+aeadSizeOverhead)
 	_, err = io.ReadFull(sc.conn, sealedFrame)
 	if err != nil {
 		return
 	}
 
-	// decrypt the frame
+	aead, err := chacha20poly1305.New(sc.recvSecret[:])
+	if err != nil {
+		return n, errors.New("Invalid SecretConnection Key")
+	}
+
+	// decrypt the frame.
+	// reads and updates the sc.recvNonce
 	var frame = make([]byte, totalFrameSize)
 	_, err = aead.Open(frame[:0], sc.recvNonce[:], sealedFrame, nil)
 	if err != nil {
@@ -172,12 +200,13 @@ func (sc *SecretConnection) Read(data []byte) (n int, err error) {
 	incrNonce(sc.recvNonce)
 	// end decryption
 
+	// copy checkLength worth into data,
+	// set recvBuffer to the rest.
 	var chunkLength = binary.LittleEndian.Uint32(frame) // read the first four bytes
 	if chunkLength > dataMaxSize {
 		return 0, errors.New("chunkLength is greater than dataMaxSize")
 	}
 	var chunk = frame[dataLenSize : dataLenSize+chunkLength]
-
 	n = copy(data, chunk)
 	sc.recvBuffer = chunk[n:]
 	return

p2p/conn/secret_connection_test.go

@@ -7,10 +7,12 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"net"
 	"os"
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -98,6 +100,69 @@ func TestSecretConnectionHandshake(t *testing.T) {
 	}
 }
 
+func TestConcurrentWrite(t *testing.T) {
+	fooSecConn, barSecConn := makeSecretConnPair(t)
+	fooWriteText := cmn.RandStr(dataMaxSize)
+
+	// write from two routines.
+	// should be safe from race according to net.Conn:
+	// https://golang.org/pkg/net/#Conn
+	n := 100
+	wg := new(sync.WaitGroup)
+	wg.Add(3)
+	go writeLots(t, wg, fooSecConn, fooWriteText, n)
+	go writeLots(t, wg, fooSecConn, fooWriteText, n)
+
+	// Consume reads from bar's reader
+	readLots(t, wg, barSecConn, n*2)
+	wg.Wait()
+
+	if err := fooSecConn.Close(); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestConcurrentRead(t *testing.T) {
+	fooSecConn, barSecConn := makeSecretConnPair(t)
+	fooWriteText := cmn.RandStr(dataMaxSize)
+	n := 100
+
+	// read from two routines.
+	// should be safe from race according to net.Conn:
+	// https://golang.org/pkg/net/#Conn
+	wg := new(sync.WaitGroup)
+	wg.Add(3)
+	go readLots(t, wg, fooSecConn, n/2)
+	go readLots(t, wg, fooSecConn, n/2)
+
+	// write to bar
+	writeLots(t, wg, barSecConn, fooWriteText, n)
+	wg.Wait()
+
+	if err := fooSecConn.Close(); err != nil {
+		t.Error(err)
+	}
+}
+
+func writeLots(t *testing.T, wg *sync.WaitGroup, conn net.Conn, txt string, n int) {
+	defer wg.Done()
+	for i := 0; i < n; i++ {
+		_, err := conn.Write([]byte(txt))
+		if err != nil {
+			t.Fatalf("Failed to write to fooSecConn: %v", err)
+		}
+	}
+}
+
+func readLots(t *testing.T, wg *sync.WaitGroup, conn net.Conn, n int) {
+	readBuffer := make([]byte, dataMaxSize)
+	for i := 0; i < n; i++ {
+		_, err := conn.Read(readBuffer)
+		assert.NoError(t, err)
+	}
+	wg.Done()
+}
+
 func TestSecretConnectionReadWrite(t *testing.T) {
 	fooConn, barConn := makeKVStoreConnPair()
 	fooWrites, barWrites := []string{}, []string{}