Merge rade/871_version_header (PR #920): protocol versioning

Closes #871.

router/connection.go

@@ -41,6 +41,7 @@ type LocalConnection struct {
 	sync.RWMutex
 	RemoteConnection
 	TCPConn           *net.TCPConn
+	version           byte
 	tcpSender         TCPSender
 	remoteUDPAddr     *net.UDPAddr
 	receivedHeartbeat bool
@@ -284,15 +285,13 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 	defer func() { conn.shutdown(err) }()
 	defer close(finished)
 
-	tcpConn := conn.TCPConn
-	tcpConn.SetLinger(0)
-	enc := gob.NewEncoder(tcpConn)
-	dec := gob.NewDecoder(tcpConn)
+	conn.TCPConn.SetLinger(0)
 
-	if err = conn.handshake(enc, dec, acceptNewPeer); err != nil {
+	_, dec, err := conn.handshake(acceptNewPeer)
+	if err != nil {
 		return
 	}
-	conn.Log("completed handshake")
+	conn.Log("completed handshake; using protocol version", conn.version)
 
 	// The ordering of the following is very important. [1]
 

router/consts.go

@@ -24,6 +24,7 @@ const (
 	PMTUVerifyTimeout   = 10 * time.Millisecond // gets doubled with every attempt
 	MaxDuration         = time.Duration(math.MaxInt64)
 	MaxMissedHeartbeats = 6
+	HeaderTimeout       = 10 * time.Second
 	HeartbeatTimeout    = MaxMissedHeartbeats * SlowHeartbeat
 )
 

router/handshake.go

@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"fmt"
 	"strconv"
+	"time"
 )
 
 type FieldValidator struct {
@@ -44,58 +45,68 @@ func (fv *FieldValidator) Err() error {
 	return fv.err
 }
 
-func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) error {
+func (conn *LocalConnection) handshake(acceptNewPeer bool) (*gob.Encoder, *gob.Decoder, error) {
 	// We do not need to worry about locking in here as at this point
 	// the connection is not reachable by any go-routine other than
 	// ourself. Only when we add this connection to the conn.local
 	// peer will it be visible from multiple go-routines.
 
+	conn.TCPConn.SetDeadline(time.Now().Add(HeaderTimeout))
+	version, err := exchangeProtocolHeader(conn.TCPConn, conn.TCPConn)
+	if err != nil {
+		return nil, nil, err
+	}
+	conn.version = version
+	conn.TCPConn.SetDeadline(time.Time{})
 	conn.extendReadDeadline()
 
 	localConnID := randUint64()
 	usingPassword := conn.Router.UsingPassword()
-	fv, private, err := conn.handshakeSendRecv(localConnID, usingPassword, enc, dec)
+	enc := gob.NewEncoder(conn.TCPConn)
+	dec := gob.NewDecoder(conn.TCPConn)
+	fv, private, err := conn.exchangeHandshake(localConnID, usingPassword, enc, dec)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
+
 	nameStr, _ := fv.Value("Name")
 	nickNameStr, _ := fv.Value("NickName")
 	uidStr, _ := fv.Value("UID")
 	remoteConnIDStr, _ := fv.Value("ConnID")
 	if err := fv.Err(); err != nil {
-		return err
+		return nil, nil, err
 	}
 
 	name, err := PeerNameFromString(nameStr)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	if !acceptNewPeer {
 		if _, found := conn.Router.Peers.Fetch(name); !found {
-			return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
+			return nil, nil, fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
 		}
 	}
 	if existingConn, found := conn.Router.Ourself.ConnectionTo(name); found && existingConn.Established() {
-		return fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr())
+		return nil, nil, fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr())
 	}
 	uid, err := ParsePeerUID(uidStr)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	remoteConnID, err := strconv.ParseUint(remoteConnIDStr, 10, 64)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	conn.uid = localConnID ^ remoteConnID
 
 	remotePublicStr, rpErr := fv.Value("PublicKey")
 	if usingPassword {
 		if rpErr != nil {
-			return rpErr
+			return nil, nil, rpErr
 		}
 		remotePublicSlice, rpErr := hex.DecodeString(remotePublicStr)
 		if rpErr != nil {
-			return rpErr
+			return nil, nil, rpErr
 		}
 		remotePublic := [32]byte{}
 		for idx, elem := range remotePublicSlice {
@@ -106,27 +117,23 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep
 		conn.Decryptor = NewNaClDecryptor(conn.SessionKey, conn.outbound)
 	} else {
 		if rpErr == nil {
-			return fmt.Errorf("Remote network is encrypted. Password required.")
+			return nil, nil, fmt.Errorf("Remote network is encrypted. Password required.")
 		}
 		conn.tcpSender = NewSimpleTCPSender(enc)
 		conn.Decryptor = NewNonDecryptor()
 	}
 
-	return conn.setRemote(NewPeer(name, nickNameStr, uid, 0))
+	return enc, dec, conn.setRemote(NewPeer(name, nickNameStr, uid, 0))
 }
 
-func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword bool, enc *gob.Encoder, dec *gob.Decoder) (*FieldValidator, *[32]byte, error) {
-	versionStr := fmt.Sprint(ProtocolVersion)
+func (conn *LocalConnection) exchangeHandshake(localConnID uint64, usingPassword bool, enc *gob.Encoder, dec *gob.Decoder) (*FieldValidator, *[32]byte, error) {
 	handshakeSend := map[string]string{
-		"Protocol":        Protocol,
-		"ProtocolVersion": versionStr,
 		"PeerNameFlavour": PeerNameFlavour,
 		"Name":            conn.local.Name.String(),
 		"NickName":        conn.local.NickName,
 		"UID":             fmt.Sprint(conn.local.UID),
 		"ConnID":          fmt.Sprint(localConnID)}
 	handshakeRecv := map[string]string{}
-
 	var public, private *[32]byte
 	var err error
 	if usingPassword {
@@ -137,14 +144,10 @@ func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword
 		handshakeSend["PublicKey"] = hex.EncodeToString(public[:])
 	}
 	enc.Encode(handshakeSend)
-
-	err = dec.Decode(&handshakeRecv)
-	if err != nil {
+	if err = dec.Decode(&handshakeRecv); err != nil {
 		return nil, nil, err
 	}
 	fv := NewFieldValidator(handshakeRecv)
-	fv.CheckEqual("Protocol", Protocol)
-	fv.CheckEqual("ProtocolVersion", versionStr)
 	fv.CheckEqual("PeerNameFlavour", PeerNameFlavour)
 	return fv, private, nil
 }

router/protocol.go

@@ -1,10 +1,59 @@
 package router
 
-const (
-	Protocol        = "weave"
-	ProtocolVersion = 20
+import (
+	"bytes"
+	"fmt"
+	"io"
 )
 
+const (
+	Protocol                = "weave"
+	ProtocolVersion    byte = 1
+	ProtocolMinVersion byte = ProtocolVersion
+)
+
+var (
+	ProtocolBytes     = []byte(Protocol)
+	ProtocolLen       = len(ProtocolBytes)
+	ProtocolHeader    = append(ProtocolBytes, ProtocolMinVersion, ProtocolVersion)
+	ProtocolHeaderLen = len(ProtocolHeader)
+)
+
+func exchangeProtocolHeader(w io.Writer, r io.Reader) (byte, error) {
+	if n, err := w.Write(ProtocolHeader); err != nil {
+		return 0, err
+	} else if n != ProtocolHeaderLen {
+		return 0, fmt.Errorf("failed to send complete protocol header")
+	}
+	header := make([]byte, ProtocolHeaderLen)
+	if n, err := io.ReadFull(r, header); err != nil && n == 0 {
+		return 0, fmt.Errorf("failed to receive remote protocol header: %s", err)
+	} else if err != nil {
+		return 0, fmt.Errorf("received incomplete remote protocol header (%d octets instead of %d): %v; error: %s",
+			n, ProtocolHeaderLen, header[:n], err)
+	}
+	if !bytes.Equal(ProtocolBytes, header[:ProtocolLen]) {
+		return 0, fmt.Errorf("remote protocol header not recognised: %v", header[:ProtocolHeaderLen])
+	}
+	var (
+		remoteMinVersion = header[ProtocolLen]
+		remoteVersion    = header[ProtocolLen+1]
+	)
+	minVersion := ProtocolMinVersion
+	if remoteMinVersion > minVersion {
+		minVersion = remoteMinVersion
+	}
+	maxVersion := ProtocolVersion
+	if remoteVersion < maxVersion {
+		maxVersion = remoteVersion
+	}
+	if minVersion > maxVersion {
+		return 0, fmt.Errorf("remote version [%d,%d] is incompatible with our version [%d,%d]",
+			remoteMinVersion, remoteVersion, ProtocolMinVersion, ProtocolVersion)
+	}
+	return maxVersion, nil
+}
+
 type ProtocolTag byte
 
 const (