kademlia: handle overlay mismatch (#356)

* handle overlay mismatch
Co-authored-by: Petar Radovic <petar.radovic@gmail.com>

pkg/kademlia/kademlia.go

@@ -29,6 +29,7 @@ const (
 
 var (
 	errMissingAddressBookEntry = errors.New("addressbook underlay entry not found")
+	errOverlayMismatch         = errors.New("overlay mismatch")
 	timeToRetry                = 60 * time.Second
 	shortRetry                 = 30 * time.Second
 )
@@ -156,6 +157,12 @@ func (k *Kad) manage() {
 
 				err = k.connect(ctx, peer, bzzAddr.Underlay, po)
 				if err != nil {
+					if errors.Is(err, errOverlayMismatch) {
+						k.knownPeers.Remove(peer, po)
+						if err := k.addressBook.Remove(peer); err != nil {
+							k.logger.Debugf("could not remove peer from addressbook: %s", peer.String())
+						}
+					}
 					k.logger.Debugf("error connecting to peer from kademlia %s: %v", bzzAddr.String(), err)
 					k.logger.Warningf("connecting to peer %s: %v", bzzAddr.ShortString(), err)
 					// continue to next
@@ -260,7 +267,7 @@ func (k *Kad) recalcDepth() uint8 {
 func (k *Kad) connect(ctx context.Context, peer swarm.Address, ma ma.Multiaddr, po uint8) error {
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
-	_, err := k.p2p.Connect(ctx, ma)
+	i, err := k.p2p.Connect(ctx, ma)
 	if err != nil {
 		if errors.Is(err, p2p.ErrAlreadyConnected) {
 			return nil
@@ -287,6 +294,7 @@ func (k *Kad) connect(ctx context.Context, peer swarm.Address, ma ma.Multiaddr,
 			if err := k.addressBook.Remove(peer); err != nil {
 				k.logger.Debugf("could not remove peer from addressbook: %s", peer.String())
 			}
+			k.logger.Debugf("kademlia pruned peer from address book %s", peer.String())
 		} else {
 			k.waitNext[peer.String()] = retryInfo{tryAfter: retryTime, failedAttempts: failedAttempts}
 		}
@@ -295,6 +303,12 @@ func (k *Kad) connect(ctx context.Context, peer swarm.Address, ma ma.Multiaddr,
 		return err
 	}
 
+	if !i.Overlay.Equal(peer) {
+		_ = k.p2p.Disconnect(peer)
+		_ = k.p2p.Disconnect(i.Overlay)
+		return errOverlayMismatch
+	}
+
 	return k.announce(ctx, peer)
 }
 

pkg/kademlia/kademlia_test.go

@@ -45,8 +45,7 @@ var nonConnectableAddress, _ = ma.NewMultiaddr(underlayBase + "16Uiu2HAkx8ULY8cT
 // tested in TestManage below.
 func TestNeighborhoodDepth(t *testing.T) {
 	var (
-		conns int32 // how many connect calls were made to the p2p mock
-
+		conns                    int32 // how many connect calls were made to the p2p mock
 		base, kad, ab, _, signer = newTestKademlia(&conns, nil, nil)
 		peers                    []swarm.Address
 		binEight                 []swarm.Address
@@ -443,7 +442,7 @@ func TestClosestPeer(t *testing.T) {
 	ab := addressbook.New(mockstate.NewStateStore())
 	var conns int32
 
-	kad := kademlia.New(kademlia.Options{Base: base, Discovery: disc, AddressBook: ab, P2P: p2pMock(&conns, nil), Logger: logger})
+	kad := kademlia.New(kademlia.Options{Base: base, Discovery: disc, AddressBook: ab, P2P: p2pMock(ab, &conns, nil), Logger: logger})
 	defer kad.Close()
 
 	pk, _ := crypto.GenerateSecp256k1Key()
@@ -613,10 +612,10 @@ func TestMarshal(t *testing.T) {
 
 func newTestKademlia(connCounter, failedConnCounter *int32, f func(bin, depth uint8, peers *pslice.PSlice) bool) (swarm.Address, *kademlia.Kad, addressbook.Interface, *mock.Discovery, beeCrypto.Signer) {
 	var (
-		base   = test.RandomAddress() // base address
-		p2p    = p2pMock(connCounter, failedConnCounter)
+		base   = test.RandomAddress()                       // base address
+		ab     = addressbook.New(mockstate.NewStateStore()) // address book
+		p2p    = p2pMock(ab, connCounter, failedConnCounter)
 		logger = logging.New(ioutil.Discard, 0)                                                                                            // logger
-		ab     = addressbook.New(mockstate.NewStateStore())                                                                                // address book
 		disc   = mock.NewDiscovery()                                                                                                       // mock discovery
 		kad    = kademlia.New(kademlia.Options{Base: base, Discovery: disc, AddressBook: ab, P2P: p2p, Logger: logger, SaturationFunc: f}) // kademlia instance
 	)
@@ -625,7 +624,7 @@ func newTestKademlia(connCounter, failedConnCounter *int32, f func(bin, depth ui
 	return base, kad, ab, disc, beeCrypto.NewDefaultSigner(pk)
 }
 
-func p2pMock(counter, failedCounter *int32) p2p.Service {
+func p2pMock(ab addressbook.Interface, counter, failedCounter *int32) p2p.Service {
 	p2ps := p2pmock.New(p2pmock.WithConnectFunc(func(ctx context.Context, addr ma.Multiaddr) (*bzz.Address, error) {
 		if addr.Equal(nonConnectableAddress) {
 			_ = atomic.AddInt32(failedCounter, 1)
@@ -634,6 +633,18 @@ func p2pMock(counter, failedCounter *int32) p2p.Service {
 		if counter != nil {
 			_ = atomic.AddInt32(counter, 1)
 		}
+
+		addresses, err := ab.Addresses()
+		if err != nil {
+			return nil, errors.New("could not fetch addresbook addresses")
+		}
+
+		for _, a := range addresses {
+			if a.Underlay.Equal(addr) {
+				return &a, nil
+			}
+		}
+
 		return nil, nil
 	}))