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)...)) +}