Merge pull request #1350 from tendermint/1275-p2p-loopbacks

p2p: loopbacks should be detected and ignored instead of dialling self infinitely

CHANGELOG.md

@@ -35,6 +35,7 @@ IMPROVEMENTS:
 - [config] trim whitespace from elements of lists (like `persistent_peers`)
 - [rpc] `/tx` and `/tx_search` responses now include the transaction hash
 - [rpc] include validator power in `/status`
+- [p2p] do not try to connect to ourselves (ok, maybe only once)
 
 BUG FIXES:
 - [rpc] fix subscribing using an abci.ResponseDeliverTx tag

node/node.go

@@ -405,9 +405,14 @@ func (n *Node) OnStart() error {
 	}
 	n.Logger.Info("P2P Node ID", "ID", nodeKey.ID(), "file", n.config.NodeKeyFile())
 
-	// Start the switch
+	nodeInfo := n.makeNodeInfo(nodeKey.PubKey())
-	n.sw.SetNodeInfo(n.makeNodeInfo(nodeKey.PubKey()))
+	n.sw.SetNodeInfo(nodeInfo)
 	n.sw.SetNodeKey(nodeKey)
+
+	// Add ourselves to addrbook to prevent dialing ourselves
+	n.addrBook.AddOurAddress(nodeInfo.NetAddress())
+
+	// Start the switch
 	err = n.sw.Start()
 	if err != nil {
 		return err

p2p/node_info.go

@@ -107,6 +107,7 @@ OUTER_LOOP:
 	return nil
 }
 
+// ID returns node's ID.
 func (info NodeInfo) ID() ID {
 	return PubKeyToID(info.PubKey)
 }

p2p/pex/addrbook.go

@@ -33,10 +33,15 @@ type AddrBook interface {
 
 	// Add our own addresses so we don't later add ourselves
 	AddOurAddress(*p2p.NetAddress)
+	// Check if it is our address
+	OurAddress(*p2p.NetAddress) bool
 
 	// Add and remove an address
 	AddAddress(addr *p2p.NetAddress, src *p2p.NetAddress) error
-	RemoveAddress(addr *p2p.NetAddress)
+	RemoveAddress(*p2p.NetAddress)
+
+	// Check if the address is in the book
+	HasAddress(*p2p.NetAddress) bool
 
 	// Do we need more peers?
 	NeedMoreAddrs() bool
@@ -78,7 +83,7 @@ type addrBook struct {
 	// accessed concurrently
 	mtx        sync.Mutex
 	rand       *rand.Rand
-	ourAddrs   map[string]*p2p.NetAddress
+	ourAddrs   map[string]struct{}
 	addrLookup map[p2p.ID]*knownAddress // new & old
 	bucketsOld []map[string]*knownAddress
 	bucketsNew []map[string]*knownAddress
@@ -93,7 +98,7 @@ type addrBook struct {
 func NewAddrBook(filePath string, routabilityStrict bool) *addrBook {
 	am := &addrBook{
 		rand:              rand.New(rand.NewSource(time.Now().UnixNano())), // TODO: seed from outside
-		ourAddrs:          make(map[string]*p2p.NetAddress),
+		ourAddrs:          make(map[string]struct{}),
 		addrLookup:        make(map[p2p.ID]*knownAddress),
 		filePath:          filePath,
 		routabilityStrict: routabilityStrict,
@@ -154,7 +159,15 @@ func (a *addrBook) AddOurAddress(addr *p2p.NetAddress) {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 	a.Logger.Info("Add our address to book", "addr", addr)
-	a.ourAddrs[addr.String()] = addr
+	a.ourAddrs[addr.String()] = struct{}{}
+}
+
+// OurAddress returns true if it is our address.
+func (a *addrBook) OurAddress(addr *p2p.NetAddress) bool {
+	a.mtx.Lock()
+	_, ok := a.ourAddrs[addr.String()]
+	a.mtx.Unlock()
+	return ok
 }
 
 // AddAddress implements AddrBook - adds the given address as received from the given source.
@@ -185,6 +198,14 @@ func (a *addrBook) IsGood(addr *p2p.NetAddress) bool {
 	return a.addrLookup[addr.ID].isOld()
 }
 
+// HasAddress returns true if the address is in the book.
+func (a *addrBook) HasAddress(addr *p2p.NetAddress) bool {
+	a.mtx.Lock()
+	defer a.mtx.Unlock()
+	ka := a.addrLookup[addr.ID]
+	return ka != nil
+}
+
 // NeedMoreAddrs implements AddrBook - returns true if there are not have enough addresses in the book.
 func (a *addrBook) NeedMoreAddrs() bool {
 	return a.Size() < needAddressThreshold

p2p/pex/addrbook_test.go

@@ -338,3 +338,19 @@ func TestAddrBookGetSelectionWithBias(t *testing.T) {
 		t.Fatalf("expected more good peers (%% got: %d, %% expected: %d, number of good addrs: %d, total: %d)", got, expected, good, len(selection))
 	}
 }
+
+func TestAddrBookHasAddress(t *testing.T) {
+	fname := createTempFileName("addrbook_test")
+	defer deleteTempFile(fname)
+
+	book := NewAddrBook(fname, true)
+	book.SetLogger(log.TestingLogger())
+	addr := randIPv4Address(t)
+	book.AddAddress(addr, addr)
+
+	assert.True(t, book.HasAddress(addr))
+
+	book.RemoveAddress(addr)
+
+	assert.False(t, book.HasAddress(addr))
+}

p2p/pex/pex_reactor.go

@@ -164,7 +164,10 @@ func (r *PEXReactor) AddPeer(p Peer) {
 		// peers when we need - we don't trust inbound peers as much.
 		addr := p.NodeInfo().NetAddress()
 		if !isAddrPrivate(addr, r.config.PrivatePeerIDs) {
-			r.book.AddAddress(addr, addr)
+			err := r.book.AddAddress(addr, addr)
+			if err != nil {
+				r.Logger.Error("Failed to add new address", "err", err)
+			}
 		}
 	}
 }
@@ -265,7 +268,10 @@ func (r *PEXReactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
 	srcAddr := src.NodeInfo().NetAddress()
 	for _, netAddr := range addrs {
 		if netAddr != nil && !isAddrPrivate(netAddr, r.config.PrivatePeerIDs) {
-			r.book.AddAddress(netAddr, srcAddr)
+			err := r.book.AddAddress(netAddr, srcAddr)
+			if err != nil {
+				r.Logger.Error("Failed to add new address", "err", err)
+			}
 		}
 	}
 	return nil

p2p/pex/pex_reactor_test.go

@@ -63,35 +63,45 @@ func TestPEXReactorRunning(t *testing.T) {
 	N := 3
 	switches := make([]*p2p.Switch, N)
 
+	// directory to store address books
 	dir, err := ioutil.TempDir("", "pex_reactor")
 	require.Nil(t, err)
 	defer os.RemoveAll(dir) // nolint: errcheck
-	book := NewAddrBook(filepath.Join(dir, "addrbook.json"), false)
+
-	book.SetLogger(log.TestingLogger())
+	books := make([]*addrBook, N)
+	logger := log.TestingLogger()
 
 	// create switches
 	for i := 0; i < N; i++ {
 		switches[i] = p2p.MakeSwitch(config, i, "127.0.0.1", "123.123.123", func(i int, sw *p2p.Switch) *p2p.Switch {
-			sw.SetLogger(log.TestingLogger().With("switch", i))
+			books[i] = NewAddrBook(filepath.Join(dir, fmt.Sprintf("addrbook%d.json", i)), false)
+			books[i].SetLogger(logger.With("pex", i))
+			sw.SetAddrBook(books[i])
 
-			r := NewPEXReactor(book, &PEXReactorConfig{})
+			sw.SetLogger(logger.With("pex", i))
-			r.SetLogger(log.TestingLogger())
+
+			r := NewPEXReactor(books[i], &PEXReactorConfig{})
+			r.SetLogger(logger.With("pex", i))
 			r.SetEnsurePeersPeriod(250 * time.Millisecond)
 			sw.AddReactor("pex", r)
+
 			return sw
 		})
 	}
 
-	// fill the address book and add listeners
+	addOtherNodeAddrToAddrBook := func(switchIndex, otherSwitchIndex int) {
-	for _, s := range switches {
+		addr := switches[otherSwitchIndex].NodeInfo().NetAddress()
-		addr := s.NodeInfo().NetAddress()
+		books[switchIndex].AddAddress(addr, addr)
-		book.AddAddress(addr, addr)
-		s.AddListener(p2p.NewDefaultListener("tcp", s.NodeInfo().ListenAddr, true, log.TestingLogger()))
 	}
 
-	// start switches
+	addOtherNodeAddrToAddrBook(0, 1)
-	for _, s := range switches {
+	addOtherNodeAddrToAddrBook(1, 0)
-		err := s.Start() // start switch and reactors
+	addOtherNodeAddrToAddrBook(2, 1)
+
+	for i, sw := range switches {
+		sw.AddListener(p2p.NewDefaultListener("tcp", sw.NodeInfo().ListenAddr, true, logger.With("pex", i)))
+
+		err := sw.Start() // start switch and reactors
 		require.Nil(t, err)
 	}
 
@@ -127,6 +137,7 @@ func TestPEXReactorRequestMessageAbuse(t *testing.T) {
 	defer teardownReactor(book)
 
 	sw := createSwitchAndAddReactors(r)
+	sw.SetAddrBook(book)
 
 	peer := newMockPeer()
 	p2p.AddPeerToSwitch(sw, peer)
@@ -156,6 +167,7 @@ func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
 	defer teardownReactor(book)
 
 	sw := createSwitchAndAddReactors(r)
+	sw.SetAddrBook(book)
 
 	peer := newMockPeer()
 	p2p.AddPeerToSwitch(sw, peer)
@@ -182,13 +194,11 @@ func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
 }
 
 func TestPEXReactorUsesSeedsIfNeeded(t *testing.T) {
+	// directory to store address books
 	dir, err := ioutil.TempDir("", "pex_reactor")
 	require.Nil(t, err)
 	defer os.RemoveAll(dir) // nolint: errcheck
 
-	book := NewAddrBook(filepath.Join(dir, "addrbook.json"), false)
-	book.SetLogger(log.TestingLogger())
-
 	// 1. create seed
 	seed := p2p.MakeSwitch(
 		config,
@@ -196,6 +206,10 @@ func TestPEXReactorUsesSeedsIfNeeded(t *testing.T) {
 		"127.0.0.1",
 		"123.123.123",
 		func(i int, sw *p2p.Switch) *p2p.Switch {
+			book := NewAddrBook(filepath.Join(dir, "addrbook0.json"), false)
+			book.SetLogger(log.TestingLogger())
+			sw.SetAddrBook(book)
+
 			sw.SetLogger(log.TestingLogger())
 
 			r := NewPEXReactor(book, &PEXReactorConfig{})
@@ -222,6 +236,10 @@ func TestPEXReactorUsesSeedsIfNeeded(t *testing.T) {
 		"127.0.0.1",
 		"123.123.123",
 		func(i int, sw *p2p.Switch) *p2p.Switch {
+			book := NewAddrBook(filepath.Join(dir, "addrbook1.json"), false)
+			book.SetLogger(log.TestingLogger())
+			sw.SetAddrBook(book)
+
 			sw.SetLogger(log.TestingLogger())
 
 			r := NewPEXReactor(
@@ -247,7 +265,8 @@ func TestPEXReactorCrawlStatus(t *testing.T) {
 	defer teardownReactor(book)
 
 	// Seed/Crawler mode uses data from the Switch
-	_ = createSwitchAndAddReactors(pexR)
+	sw := createSwitchAndAddReactors(pexR)
+	sw.SetAddrBook(book)
 
 	// Create a peer, add it to the peer set and the addrbook.
 	peer := p2p.CreateRandomPeer(false)
@@ -291,7 +310,8 @@ func TestPEXReactorDialPeer(t *testing.T) {
 	pexR, book := createReactor(&PEXReactorConfig{})
 	defer teardownReactor(book)
 
-	_ = createSwitchAndAddReactors(pexR)
+	sw := createSwitchAndAddReactors(pexR)
+	sw.SetAddrBook(book)
 
 	peer := newMockPeer()
 	addr := peer.NodeInfo().NetAddress()
@@ -397,6 +417,7 @@ func assertPeersWithTimeout(
 }
 
 func createReactor(config *PEXReactorConfig) (r *PEXReactor, book *addrBook) {
+	// directory to store address book
 	dir, err := ioutil.TempDir("", "pex_reactor")
 	if err != nil {
 		panic(err)

p2p/switch.go

@@ -33,15 +33,21 @@ const (
 
 //-----------------------------------------------------------------------------
 
+// An AddrBook represents an address book from the pex package, which is used
+// to store peer addresses.
 type AddrBook interface {
 	AddAddress(addr *NetAddress, src *NetAddress) error
+	AddOurAddress(*NetAddress)
+	OurAddress(*NetAddress) bool
 	MarkGood(*NetAddress)
+	RemoveAddress(*NetAddress)
+	HasAddress(*NetAddress) bool
 	Save()
 }
 
 //-----------------------------------------------------------------------------
 
-// `Switch` handles peer connections and exposes an API to receive incoming messages
+// Switch handles peer connections and exposes an API to receive incoming messages
 // on `Reactors`.  Each `Reactor` is responsible for handling incoming messages of one
 // or more `Channels`.  So while sending outgoing messages is typically performed on the peer,
 // incoming messages are received on the reactor.
@@ -66,6 +72,7 @@ type Switch struct {
 	rng *rand.Rand // seed for randomizing dial times and orders
 }
 
+// NewSwitch creates a new Switch with the given config.
 func NewSwitch(config *cfg.P2PConfig) *Switch {
 	sw := &Switch{
 		config:       config,
@@ -343,20 +350,21 @@ func (sw *Switch) IsDialing(id ID) bool {
 // DialPeersAsync dials a list of peers asynchronously in random order (optionally, making them persistent).
 func (sw *Switch) DialPeersAsync(addrBook AddrBook, peers []string, persistent bool) error {
 	netAddrs, errs := NewNetAddressStrings(peers)
+	// only log errors, dial correct addresses
 	for _, err := range errs {
 		sw.Logger.Error("Error in peer's address", "err", err)
 	}
 
+	ourAddr := sw.nodeInfo.NetAddress()
+
+	// TODO: move this out of here ?
 	if addrBook != nil {
 		// add peers to `addrBook`
-		ourAddr := sw.nodeInfo.NetAddress()
 		for _, netAddr := range netAddrs {
 			// do not add our address or ID
-			if netAddr.Same(ourAddr) {
+			if !netAddr.Same(ourAddr) {
-				continue
+				addrBook.AddAddress(netAddr, ourAddr)
 			}
-			// TODO: move this out of here ?
-			addrBook.AddAddress(netAddr, ourAddr)
 		}
 		// Persist some peers to disk right away.
 		// NOTE: integration tests depend on this
@@ -367,8 +375,14 @@ func (sw *Switch) DialPeersAsync(addrBook AddrBook, peers []string, persistent b
 	perm := sw.rng.Perm(len(netAddrs))
 	for i := 0; i < len(perm); i++ {
 		go func(i int) {
-			sw.randomSleep(0)
 			j := perm[i]
+
+			// do not dial ourselves
+			if netAddrs[j].Same(ourAddr) {
+				return
+			}
+
+			sw.randomSleep(0)
 			err := sw.DialPeerWithAddress(netAddrs[j], persistent)
 			if err != nil {
 				sw.Logger.Error("Error dialing peer", "err", err)
@@ -522,6 +536,15 @@ func (sw *Switch) addPeer(pc peerConn) error {
 
 	// Avoid self
 	if sw.nodeKey.ID() == peerID {
+		addr := peerNodeInfo.NetAddress()
+
+		// remove the given address from the address book if we're added it earlier
+		sw.addrBook.RemoveAddress(addr)
+
+		// add the given address to the address book to avoid dialing ourselves
+		// again this is our public address
+		sw.addrBook.AddOurAddress(addr)
+
 		return ErrSwitchConnectToSelf
 	}
 

p2p/switch_test.go

@@ -39,8 +39,6 @@ type TestReactor struct {
 
 	mtx          sync.Mutex
 	channels     []*conn.ChannelDescriptor
-	peersAdded   []Peer
-	peersRemoved []Peer
 	logMessages  bool
 	msgsCounter  int
 	msgsReceived map[byte][]PeerMessage
@@ -61,17 +59,9 @@ func (tr *TestReactor) GetChannels() []*conn.ChannelDescriptor {
 	return tr.channels
 }
 
-func (tr *TestReactor) AddPeer(peer Peer) {
+func (tr *TestReactor) AddPeer(peer Peer) {}
-	tr.mtx.Lock()
-	defer tr.mtx.Unlock()
-	tr.peersAdded = append(tr.peersAdded, peer)
-}
 
-func (tr *TestReactor) RemovePeer(peer Peer, reason interface{}) {
+func (tr *TestReactor) RemovePeer(peer Peer, reason interface{}) {}
-	tr.mtx.Lock()
-	defer tr.mtx.Unlock()
-	tr.peersRemoved = append(tr.peersRemoved, peer)
-}
 
 func (tr *TestReactor) Receive(chID byte, peer Peer, msgBytes []byte) {
 	if tr.logMessages {
@@ -100,6 +90,10 @@ func MakeSwitchPair(t testing.TB, initSwitch func(int, *Switch) *Switch) (*Switc
 }
 
 func initSwitchFunc(i int, sw *Switch) *Switch {
+	sw.SetAddrBook(&addrBookMock{
+		addrs:    make(map[string]struct{}),
+		ourAddrs: make(map[string]struct{})})
+
 	// Make two reactors of two channels each
 	sw.AddReactor("foo", NewTestReactor([]*conn.ChannelDescriptor{
 		{ID: byte(0x00), Priority: 10},
@@ -109,6 +103,7 @@ func initSwitchFunc(i int, sw *Switch) *Switch {
 		{ID: byte(0x02), Priority: 10},
 		{ID: byte(0x03), Priority: 10},
 	}, true))
+
 	return sw
 }
 
@@ -185,6 +180,32 @@ func TestConnAddrFilter(t *testing.T) {
 	assertNoPeersAfterTimeout(t, s2, 400*time.Millisecond)
 }
 
+func TestSwitchFiltersOutItself(t *testing.T) {
+	s1 := MakeSwitch(config, 1, "127.0.0.2", "123.123.123", initSwitchFunc)
+	// addr := s1.NodeInfo().NetAddress()
+
+	// // add ourselves like we do in node.go#427
+	// s1.addrBook.AddOurAddress(addr)
+
+	// simulate s1 having a public IP by creating a remote peer with the same ID
+	rp := &remotePeer{PrivKey: s1.nodeKey.PrivKey, Config: DefaultPeerConfig()}
+	rp.Start()
+
+	// addr should be rejected in addPeer based on the same ID
+	err := s1.DialPeerWithAddress(rp.Addr(), false)
+	if assert.Error(t, err) {
+		assert.Equal(t, ErrSwitchConnectToSelf, err)
+	}
+
+	assert.True(t, s1.addrBook.OurAddress(rp.Addr()))
+
+	assert.False(t, s1.addrBook.HasAddress(rp.Addr()))
+
+	rp.Stop()
+
+	assertNoPeersAfterTimeout(t, s1, 100*time.Millisecond)
+}
+
 func assertNoPeersAfterTimeout(t *testing.T, sw *Switch, timeout time.Duration) {
 	time.Sleep(timeout)
 	if sw.Peers().Size() != 0 {
@@ -350,3 +371,29 @@ func BenchmarkSwitchBroadcast(b *testing.B) {
 
 	b.Logf("success: %v, failure: %v", numSuccess, numFailure)
 }
+
+type addrBookMock struct {
+	addrs    map[string]struct{}
+	ourAddrs map[string]struct{}
+}
+
+var _ AddrBook = (*addrBookMock)(nil)
+
+func (book *addrBookMock) AddAddress(addr *NetAddress, src *NetAddress) error {
+	book.addrs[addr.String()] = struct{}{}
+	return nil
+}
+func (book *addrBookMock) AddOurAddress(addr *NetAddress) { book.ourAddrs[addr.String()] = struct{}{} }
+func (book *addrBookMock) OurAddress(addr *NetAddress) bool {
+	_, ok := book.ourAddrs[addr.String()]
+	return ok
+}
+func (book *addrBookMock) MarkGood(*NetAddress) {}
+func (book *addrBookMock) HasAddress(addr *NetAddress) bool {
+	_, ok := book.addrs[addr.String()]
+	return ok
+}
+func (book *addrBookMock) RemoveAddress(addr *NetAddress) {
+	delete(book.addrs, addr.String())
+}
+func (book *addrBookMock) Save() {}