more forward/backward compatible protocol versioning
We now first exchange a very minimal header, consisting of just the
name of the protocol ("weave") and the min and max version (encoded as
a byte each) we support. This leaves all other details of the
protocol, including the contents of of the handshake, open.
Peers agree to communicate using the maximum mutually supported
version. We record this in the LocalConnection, though atm nothing is
making use of that information.
We set a shorter than normal deadline (10s instead of 60s) for the
header exchange, so we detect version mismatches, garbage, or
non-responsive remotes quickly.
We reset the protocol version to '1' since this is a fresh start; old
weaves will not be able to parse the header.
Closes #871.
This commit is contained in:
1 parent
0206434f62
commit
3124ffede5
4 files changed
+84
-32
No files matched your search
@@ -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]
|
||||
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
+26
-23
@@ -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
|
||||
}
|
||||
|
||||
+52
-3
@@ -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 (
|
||||
|
||||
Reference in new issue
Block a user