From f522f5b730a79e6f2cdffc9277c879679e13e553 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Thu, 11 Jun 2015 13:58:54 +0100 Subject: [PATCH] Ensure we send a gossip update whenever a new connection requires it Previously, if LocalConnection run() failed before calling AddConnection, it would not gossip anything. This was benign, because the peer information we learn in the handshake is just a placeholder until the full information arrived via gossip, and if a connection is not formed, there is nothing else to gossip about. With short ids, there's a corner case where this is no longer true: If a new peer connects to us, and it has the same short id as we do, then we might change our short id. If we do, then it is important that we gossip that out, even if the attempt to establish the connection fails further along. --- router/connection.go | 69 ++++++++++++++++++++++++++++++++----------- router/gossip_test.go | 1 + router/handshake.go | 41 ++++++++----------------- router/local_peer.go | 10 ++----- router/router.go | 5 ++++ 5 files changed, 72 insertions(+), 54 deletions(-) diff --git a/router/connection.go b/router/connection.go index 6820039d..6794f2aa 100644 --- a/router/connection.go +++ b/router/connection.go @@ -171,24 +171,8 @@ 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) - - if err = conn.handshake(enc, dec, acceptNewPeer); err != nil { - return - } - conn.Log("completed handshake") - - // The ordering of the following is very important. [1] - - localIP := tcpConn.LocalAddr().(*net.TCPAddr).IP - if conn.forwarder, err = conn.Router.InterHost.MakeForwarder(conn.remote, localIP, conn.remoteUDPAddr, conn.uid, conn.udpCrypto(), conn.sendInterHostControlMessage); err != nil { - return - } - - if err = conn.Router.Ourself.AddConnection(conn); err != nil { + dec, err := conn.prepareConnection(acceptNewPeer) + if err != nil { return } @@ -200,6 +184,41 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch err = conn.actorLoop(actionChan) } +func (conn *LocalConnection) prepareConnection(acceptNewPeer bool) (*gob.Decoder, error) { + tcpConn := conn.TCPConn + tcpConn.SetLinger(0) + enc := gob.NewEncoder(tcpConn) + dec := gob.NewDecoder(tcpConn) + + remotePeer, err := conn.handshake(enc, dec, acceptNewPeer) + if err != nil { + return nil, err + } + conn.Log("completed handshake") + + if err = conn.setRemote(remotePeer); err != nil { + return nil, err + } + + // setRemote may have resulted in the local short id changing. + // So we need to ensure we gossip that, even if we error out + // below. + defer conn.Router.broadcastPeerUpdate(remotePeer) + + // The ordering of the following is very important. [1] + + localIP := tcpConn.LocalAddr().(*net.TCPAddr).IP + if conn.forwarder, err = conn.Router.InterHost.MakeForwarder(conn.remote, localIP, conn.remoteUDPAddr, conn.uid, conn.udpCrypto(), conn.sendInterHostControlMessage); err != nil { + return nil, err + } + + if err = conn.Router.Ourself.AddConnection(conn); err != nil { + return nil, err + } + + return dec, nil +} + // [1] Ordering constraints: // // (a) AddConnections must precede initHeartbeats. It is only after @@ -245,6 +264,20 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch // prevent that completely, since, for example, forwarder can only be // created when we know the remote UDP address, but it helps to try. +func (conn *LocalConnection) setRemote(toPeer *Peer) error { + toPeer = conn.Router.Peers.FetchWithDefault(toPeer) + switch toPeer { + case nil: + return fmt.Errorf("multiple peers with the same Name") + case conn.local: + conn.remote = toPeer // have to do assigment here to ensure Shutdown releases ref count + return fmt.Errorf("Cannot connect to ourself") + default: + conn.remote = toPeer + return nil + } +} + func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { for err == nil { select { diff --git a/router/gossip_test.go b/router/gossip_test.go index eba7a0c9..42233bee 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -38,6 +38,7 @@ func (router *Router) AddTestChannelConnection(r *Router) { conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r} router.Ourself.handleAddConnection(conn) + router.broadcastPeerUpdate(conn.Remote()) router.Ourself.handleConnectionEstablished(conn) router.sendPendingGossip() } diff --git a/router/handshake.go b/router/handshake.go index fbfc5fe3..768def0e 100644 --- a/router/handshake.go +++ b/router/handshake.go @@ -44,7 +44,7 @@ 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(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) (*Peer, 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 @@ -56,7 +56,7 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep usingPassword := conn.Router.UsingPassword() fv, private, err := conn.handshakeSendRecv(localConnID, usingPassword, enc, dec) if err != nil { - return err + return nil, err } nameStr, _ := fv.Value("Name") nickNameStr, _ := fv.Value("NickName") @@ -64,48 +64,48 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep shortIDStr, _ := fv.Value("ShortID") remoteConnIDStr, _ := fv.Value("ConnID") if err := fv.Err(); err != nil { - return err + return nil, err } name, err := PeerNameFromString(nameStr) if err != nil { - return err + return 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, 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, fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr()) } uid, err := ParsePeerUID(uidStr) if err != nil { - return err + return nil, err } shortID, err := strconv.ParseUint(shortIDStr, 10, PeerShortIDBits) if err != nil { - return err + return nil, err } remoteConnID, err := strconv.ParseUint(remoteConnIDStr, 10, 64) if err != nil { - return err + return nil, err } conn.uid = localConnID ^ remoteConnID remotePublicStr, rpErr := fv.Value("PublicKey") if usingPassword { if rpErr != nil { - return rpErr + return nil, rpErr } remotePublicSlice, rpErr := hex.DecodeString(remotePublicStr) if rpErr != nil { - return rpErr + return nil, rpErr } remotePublic := [32]byte{} for idx, elem := range remotePublicSlice { @@ -115,13 +115,12 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep conn.tcpSender = NewEncryptedTCPSender(enc, conn.SessionKey, conn.outbound) } else { if rpErr == nil { - return fmt.Errorf("Remote network is encrypted. Password required.") + return nil, fmt.Errorf("Remote network is encrypted. Password required.") } conn.tcpSender = NewSimpleTCPSender(enc) } - return conn.setRemote(NewPeer(name, nickNameStr, uid, 0, - PeerShortID(shortID))) + return NewPeer(name, nickNameStr, uid, 0, PeerShortID(shortID)), nil } func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword bool, enc *gob.Encoder, dec *gob.Decoder) (*FieldValidator, *[32]byte, error) { @@ -158,17 +157,3 @@ func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword fv.CheckEqual("PeerNameFlavour", PeerNameFlavour) return fv, private, nil } - -func (conn *LocalConnection) setRemote(toPeer *Peer) error { - toPeer = conn.Router.Peers.FetchWithDefault(toPeer) - switch toPeer { - case nil: - return fmt.Errorf("Connection appears to be with different version of a peer we already know of") - case conn.local: - conn.remote = toPeer // have to do assigment here to ensure Shutdown releases ref count - return fmt.Errorf("Cannot connect to ourself") - default: - conn.remote = toPeer - return nil - } -} diff --git a/router/local_peer.go b/router/local_peer.go index 20975313..4692eedf 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -218,7 +218,6 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) error { conn.Log("connection added (new peer)") peer.router.SendAllGossipDown(conn) } - peer.broadcastPeerUpdate(conn.Remote()) return nil } @@ -232,7 +231,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) { } peer.connectionEstablished(conn) conn.Log("connection fully established") - peer.broadcastPeerUpdate() + peer.router.broadcastPeerUpdate() } func (peer *LocalPeer) handleDeleteConnection(conn Connection) { @@ -251,16 +250,11 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) { // Must do garbage collection first to ensure we don't send out an // update with unreachable peers (can cause looping) peer.router.Peers.GarbageCollect() - peer.broadcastPeerUpdate() + peer.router.broadcastPeerUpdate() } // helpers -func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { - peer.router.Routes.Recalculate() - peer.router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.router.Peers, append(peers, peer.Peer)...)) -} - func (peer *LocalPeer) checkConnectionLimit() error { limit := peer.router.ConnLimit if 0 != limit && peer.connectionCount() >= limit { diff --git a/router/router.go b/router/router.go index 9fb4d336..38c3dd95 100644 --- a/router/router.go +++ b/router/router.go @@ -264,3 +264,8 @@ func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, PeerNameS } return origUpdate, newUpdate, nil } + +func (router *Router) broadcastPeerUpdate(peers ...*Peer) { + router.Routes.Recalculate() + router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(router.Peers, append(peers, router.Ourself.Peer)...)) +}