Huge change: Moved from long-lived and reused TLS connections inside pluto system to on-demand per client connection based ones. This should allow for proper concurrent access and also it should tremendously increase pluto's overall resilience.

Author: numbleroot

auth/postgres.go

@@ -106,9 +106,9 @@ func (p *PostgresAuthenticator) AuthenticatePlain(username string, password stri
 		// Check what type of error we received.
 		if err == pgx.ErrNoRows {
 			return -1, "", fmt.Errorf("username not found in users table or password wrong")
-		} else {
-			return -1, "", fmt.Errorf("error while trying to locate user: %s", err.Error())
 		}
+
+		return -1, "", fmt.Errorf("error while trying to locate user: %s", err.Error())
 	}
 
 	// Build the deterministic client-specific session identifier.

comm/connection.go

@@ -1,10 +1,8 @@
 package comm
 
 import (
-	"bufio"
 	"fmt"
 	"log"
-	"strings"
 	"time"
 
 	"crypto/tls"
@@ -15,56 +13,50 @@ import (
 // ReliableConnect attempts to connect to defined remote node
 // as longs as the error from previous attempts is possible
 // to be dealt with.
-func ReliableConnect(name string, remoteName string, remoteIP string, remotePort string, tlsConfig *tls.Config, retry int) (*tls.Conn, error) {
+func ReliableConnect(remoteName string, remoteAddr string, tlsConfig *tls.Config, retry int) (*tls.Conn, error) {
 
 	var err error
 	var c *tls.Conn
 
-	// Define address we are trying to connect to.
-	nodeAddr := fmt.Sprintf("%s:%s", remoteIP, remotePort)
-
 	// Define what an error looks like we can deal with.
-	okError := fmt.Sprintf("dial tcp %s: getsockopt: connection refused", nodeAddr)
+	okError := fmt.Sprintf("dial tcp %s: getsockopt: connection refused", remoteAddr)
 
 	// Initially, set error string to the one we can deal with.
 	err = fmt.Errorf(okError)
 
 	for err != nil {
 
 		// Attempt to connect to worker node.
-		c, err = tls.Dial("tcp", nodeAddr, tlsConfig)
+		c, err = tls.Dial("tcp", remoteAddr, tlsConfig)
 		if err != nil {
 
 			if err.Error() == okError {
 				time.Sleep(time.Duration(retry) * time.Millisecond)
 			} else {
-				return nil, fmt.Errorf("%s: Could not connect to port of node '%s' because of: %s\n", name, remoteName, err.Error())
+				return nil, fmt.Errorf("Could not connect to port of node '%s' because of: %s\n", remoteName, err.Error())
 			}
 		}
 	}
 
-	log.Printf("%s: Successfully connected to worker node '%s'.\n", name, remoteName)
+	log.Printf("Successfully connected to worker node '%s'.\n", remoteName)
 
 	return c, nil
 }
 
 // ReliableSend sends text to other node specified and
 // tries to reconnect in case of simple disconnects.
-func ReliableSend(conn *tls.Conn, text string, name string, remoteName string, remoteIP string, remotePort string, tlsConfig *tls.Config, timeout int, retry int) (*tls.Conn, bool, error) {
+func ReliableSend(conn *tls.Conn, text string, remoteName string, remoteAddr string, tlsConfig *tls.Config, timeout int, retry int) error {
 
 	var err error
 	var replacedConn *tls.Conn
 
-	// Track if we replaced the connection.
-	replaced := false
-
 	// Set configured timeout on waiting for response.
 	conn.SetWriteDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
 
 	// Test long-lived connection.
 	_, err = conn.Write([]byte("> ping <\r\n"))
 	if err != nil {
-		return nil, false, fmt.Errorf("sending ping to node '%s' failed with: %s\n", remoteName, err.Error())
+		return fmt.Errorf("sending ping to node '%s' failed with: %s\n", remoteName, err.Error())
 	}
 
 	// Wait for configured time to pass.
@@ -85,75 +77,17 @@ func ReliableSend(conn *tls.Conn, text string, name string, remoteName string, r
 		if err.Error() == okError {
 
 			// Connection was lost. Reconnect.
-			replacedConn, err = ReliableConnect(name, remoteName, remoteIP, remotePort, tlsConfig, retry)
+			replacedConn, err = ReliableConnect(remoteName, remoteAddr, tlsConfig, retry)
 			if err != nil {
-				return nil, false, fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
+				return fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
 			}
 
-			// Indicate we replaced connection.
-			replaced = true
-
-			// Wait configured time before attempting next transfer.
-			time.Sleep(time.Duration(retry) * time.Millisecond)
-
 			// Retry transfer.
 			_, err = fmt.Fprintf(replacedConn, "%s\r\n", text)
 		} else {
-			return nil, false, fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
+			return fmt.Errorf("could not reestablish connection with '%s': %s\n", remoteName, err.Error())
 		}
 	}
 
-	if replaced {
-		return replacedConn, replaced, nil
-	}
-
-	return conn, replaced, nil
-}
-
-// InternalSend is used by nodes of the pluto system to
-// successfully transmit a message to another node or
-// fail definitely. This prevents further handler advancement
-// in case a link failed.
-func InternalSend(conn *tls.Conn, text string, name string, remoteName string) error {
-
-	// Test long-lived connection.
-	_, err := conn.Write([]byte("> ping <\r\n"))
-	if err != nil {
-		return fmt.Errorf("%s: sending ping to node '%s' failed: %s\n", name, remoteName, err.Error())
-	}
-
-	// Write message to TLS connections.
-	_, err = fmt.Fprintf(conn, "%s\r\n", text)
-	for err != nil {
-		return fmt.Errorf("%s: sending message to node '%s' failed: %s\n", name, remoteName, err.Error())
-	}
-
 	return nil
 }
-
-// InternalReceive is used by nodes in the pluto system
-// receive an incoming message and filter out all prior
-// received ping message.
-func InternalReceive(reader *bufio.Reader) (string, error) {
-
-	var err error
-
-	// Initial value for received message in order
-	// to skip past the mandatory ping message.
-	text := "> ping <\r\n"
-
-	for text == "> ping <\r\n" {
-
-		text, err = reader.ReadString('\n')
-		if err != nil {
-			break
-		}
-	}
-
-	// If an error happened, return it.
-	if err != nil {
-		return "", err
-	}
-
-	return strings.TrimRight(text, "\r\n"), nil
-}

comm/receiver-sender_test.go

@@ -112,25 +112,13 @@ func TestSenderReceiver(t *testing.T) {
 	// Wait shortly for goroutines to have started.
 	time.Sleep(200 * time.Millisecond)
 
-	// Connect via TLS from worker-1 to storage.
-	cToN2, err := tls.Dial("tcp", fmt.Sprintf("%s:%s", testEnv.Config.Storage.IP, testEnv.Config.Storage.SyncPort), internalTLSConfigN1)
-	if err != nil {
-		t.Fatalf("[comm_test.TestSenderReceiver] Expected to be able to connect from worker-1 to storage but received: %s\n", err.Error())
-	}
-
 	// Create map of connections for worker-1.
-	connsN1 := make(map[string]*tls.Conn)
-	connsN1[n2] = cToN2
-
-	// Connect via TLS from storage to worker-1.
-	cToN1, err := tls.Dial("tcp", fmt.Sprintf("%s:%s", testEnv.Config.Workers[n1].IP, testEnv.Config.Workers[n1].SyncPort), internalTLSConfigN2)
-	if err != nil {
-		t.Fatalf("[comm_test.TestSenderReceiver] Expected to be able to connect from storage to worker-1 but received: %s\n", err.Error())
-	}
+	connsN1 := make(map[string]string)
+	connsN1[n2] = fmt.Sprintf("%s:%s", testEnv.Config.Storage.IP, testEnv.Config.Storage.SyncPort)
 
 	// Create map of connections for storage.
-	connsN2 := make(map[string]*tls.Conn)
-	connsN2[n1] = cToN1
+	connsN2 := make(map[string]string)
+	connsN2[n1] = fmt.Sprintf("%s:%s", testEnv.Config.Workers[n1].IP, testEnv.Config.Workers[n1].SyncPort)
 
 	// Initialize sending interface for worker-1.
 	chan1, err := comm.InitSender(n1, "../test-comm-sending-worker-1.log", internalTLSConfigN1, testEnv.Config.IntlConnTimeout, testEnv.Config.IntlConnRetry, chanIncN1, chanUpdN1, n1DownSender, connsN1)

comm/sender.go

@@ -28,7 +28,7 @@ type Sender struct {
 	updLog          *os.File
 	incVClock       chan string
 	updVClock       chan map[string]int
-	nodes           map[string]*tls.Conn
+	nodes           map[string]string
 	wg              *sync.WaitGroup
 	shutdown        chan struct{}
 }
@@ -40,7 +40,7 @@ type Sender struct {
 // with. It returns a channel local processes can put
 // CRDT changes into, so that those changes will be
 // communicated to connected nodes.
-func InitSender(name string, logFilePath string, tlsConfig *tls.Config, timeout int, retry int, incVClock chan string, updVClock chan map[string]int, downSender chan struct{}, nodes map[string]*tls.Conn) (chan string, error) {
+func InitSender(name string, logFilePath string, tlsConfig *tls.Config, timeout int, retry int, incVClock chan string, updVClock chan map[string]int, downSender chan struct{}, nodes map[string]string) (chan string, error) {
 
 	// Create and initialize what we need for
 	// a CRDT sender routine.
@@ -282,25 +282,19 @@ func (sender *Sender) SendMsgs() {
 				// Unlock mutex.
 				sender.lock.Unlock()
 
-				for i, conn := range sender.nodes {
+				for nodeName, nodeAddr := range sender.nodes {
 
-					// Extract address to reconnect to.
-					addrParts := strings.Split(conn.RemoteAddr().String(), ":")
+					// Connect to node.
+					conn, err := ReliableConnect(nodeName, nodeAddr, sender.tlsConfig, sender.intlConnRetry)
+					if err != nil {
+						log.Fatalf("[comm.SendMsgs] Failed to connect to %s: %s\n", err.Error())
+					}
 
 					// Send payload reliably to other nodes.
-					conn, replaced, err := ReliableSend(conn, marshalledMsg, sender.name, i, addrParts[0], addrParts[1], sender.tlsConfig, sender.intlConnTimeout, sender.intlConnRetry)
+					err = ReliableSend(conn, marshalledMsg, nodeName, nodeAddr, sender.tlsConfig, sender.intlConnTimeout, sender.intlConnRetry)
 					if err != nil {
 						log.Fatalf("[comm.SendMsgs] Failed to send: %s\n", err.Error())
 					}
-
-					if replaced {
-
-						// If connection had to be re-established, lock
-						// sender and exchange old connection with new one.
-						sender.lock.Lock()
-						sender.nodes[i] = conn
-						sender.lock.Unlock()
-					}
 				}
 
 				// Lock mutex.

crypto/config.go

@@ -22,17 +22,17 @@ func NewPublicTLSConfig(certPath string, keyPath string) (*tls.Config, error) {
 	// Good parts of them were taken from the excellent post:
 	// "Achieving a Perfect SSL Labs Score with Go":
 	// https://blog.bracelab.com/achieving-perfect-ssl-labs-score-with-go
+	// With further optimizations for speed and security from here:
+	// "So you want to expose Go on the Internet"
+	// https://blog.gopheracademy.com/advent-2016/exposing-go-on-the-internet/
 	config := &tls.Config{
 		Certificates:             make([]tls.Certificate, 1),
 		InsecureSkipVerify:       false,
 		MinVersion:               tls.VersionTLS12,
-		CurvePreferences:         []tls.CurveID{tls.CurveP521, tls.CurveP384, tls.CurveP256},
+		CurvePreferences:         []tls.CurveID{tls.CurveP256},
 		PreferServerCipherSuites: true,
 		CipherSuites: []uint16{
 			tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_RSA_WITH_AES_256_CBC_SHA,
 		},
 	}
 
@@ -62,20 +62,21 @@ func NewInternalTLSConfig(certPath string, keyPath string, rootCertPath string)
 	// Good parts of them were taken from the excellent post:
 	// "Achieving a Perfect SSL Labs Score with Go":
 	// https://blog.bracelab.com/achieving-perfect-ssl-labs-score-with-go
+	// With further optimizations for speed and security from here:
+	// "So you want to expose Go on the Internet"
+	// https://blog.gopheracademy.com/advent-2016/exposing-go-on-the-internet/
 	config := &tls.Config{
 		RootCAs:                  x509.NewCertPool(),
 		ClientCAs:                x509.NewCertPool(),
 		ClientAuth:               tls.RequireAndVerifyClientCert,
 		Certificates:             make([]tls.Certificate, 1),
 		InsecureSkipVerify:       false,
 		MinVersion:               tls.VersionTLS12,
-		CurvePreferences:         []tls.CurveID{tls.CurveP521, tls.CurveP384, tls.CurveP256},
+		CurvePreferences:         []tls.CurveID{tls.CurveP256},
 		PreferServerCipherSuites: true,
+		SessionTicketsDisabled:   false,
 		CipherSuites: []uint16{
 			tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_RSA_WITH_AES_256_CBC_SHA,
 		},
 	}