Repository: libp2p/go-libp2p
Pull request #2007 (2023-01-23 09:01:03)

Clean addresses with peer id before adding to addrbook
Fixes: https://github.com/libp2p/go-libp2p/issues/2001

sukunrt: I'm cleaning the address before adding it to addressbook. The other option is to change the check in all transports. This seemed cleaner to me. 

Patch:
diff --git a/p2p/host/peerstore/pstoreds/addr_book.go b/p2p/host/peerstore/pstoreds/addr_book.go
index 33ff1f5e..a16053b5 100644
--- a/p2p/host/peerstore/pstoreds/addr_book.go
+++ b/p2p/host/peerstore/pstoreds/addr_book.go
@@ -276,7 +276,7 @@ func (ab *dsAddrBook) AddAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Duratio
 	if ttl <= 0 {
 		return
 	}
-	addrs = cleanAddrs(addrs)
+	addrs = cleanAddrs(addrs, p)
 	ab.setAddrs(p, addrs, ttl, ttlExtend, false)
 }
 
@@ -302,7 +302,7 @@ func (ab *dsAddrBook) ConsumePeerRecord(recordEnvelope *record.Envelope, ttl tim
 		return false, nil
 	}
 
-	addrs := cleanAddrs(rec.Addrs)
+	addrs := cleanAddrs(rec.Addrs, rec.PeerID)
 	err = ab.setAddrs(rec.PeerID, addrs, ttl, ttlExtend, true)
 	if err != nil {
 		return false, err
@@ -385,7 +385,7 @@ func (ab *dsAddrBook) SetAddr(p peer.ID, addr ma.Multiaddr, ttl time.Duration) {
 
 // SetAddrs will add or update the TTLs of addresses in the AddrBook.
 func (ab *dsAddrBook) SetAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Duration) {
-	addrs = cleanAddrs(addrs)
+	addrs = cleanAddrs(addrs, p)
 	if ttl <= 0 {
 		ab.deleteAddrs(p, addrs)
 		return
@@ -598,10 +598,17 @@ func (ab *dsAddrBook) deleteAddrs(p peer.ID, addrs []ma.Multiaddr) (err error) {
 	return pr.flush(ab.ds)
 }
 
-func cleanAddrs(addrs []ma.Multiaddr) []ma.Multiaddr {
+func cleanAddrs(addrs []ma.Multiaddr, pid peer.ID) []ma.Multiaddr {
 	clean := make([]ma.Multiaddr, 0, len(addrs))
 	for _, addr := range addrs {
+		// Remove suffix of /p2p/peer-id from address
+		addr, addrPid := peer.SplitAddr(addr)
 		if addr == nil {
+			log.Warnw("Was passed a nil multiaddr", "peer", pid)
+			continue
+		}
+		if addrPid != "" && addrPid != pid {
+			log.Warnf("Was passed p2p address with a different peerId. found: %s, expected: %s", addrPid, pid)
 			continue
 		}
 		clean = append(clean, addr)
diff --git a/p2p/host/peerstore/pstoremem/addr_book.go b/p2p/host/peerstore/pstoremem/addr_book.go
index fc4fd067..67f9f914 100644
--- a/p2p/host/peerstore/pstoremem/addr_book.go
+++ b/p2p/host/peerstore/pstoremem/addr_book.go
@@ -238,11 +238,16 @@ func (mab *memoryAddrBook) addAddrsUnlocked(s *addrSegment, p peer.ID, addrs []m
 
 	exp := mab.clock.Now().Add(ttl)
 	for _, addr := range addrs {
+		// Remove suffix of /p2p/peer-id from address
+		addr, addrPid := peer.SplitAddr(addr)
 		if addr == nil {
-			log.Warnw("was passed nil multiaddr", "peer", p)
+			log.Warnw("Was passed nil multiaddr", "peer", p)
+			continue
+		}
+		if addrPid != "" && addrPid != p {
+			log.Warnf("Was passed p2p address with a different peerId. found: %s, expected: %s", addrPid, p)
 			continue
 		}
-
 		// find the highest TTL and Expiry time between
 		// existing records and function args
 		a, found := amap[string(addr.Bytes())] // won't allocate.
@@ -283,10 +288,15 @@ func (mab *memoryAddrBook) SetAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Du
 
 	exp := mab.clock.Now().Add(ttl)
 	for _, addr := range addrs {
+		addr, addrPid := peer.SplitAddr(addr)
 		if addr == nil {
 			log.Warnw("was passed nil multiaddr", "peer", p)
 			continue
 		}
+		if addrPid != "" && addrPid != p {
+			log.Warnf("was passed p2p address with a different peerId, found: %s wanted: %s", addrPid, p)
+			continue
+		}
 		aBytes := addr.Bytes()
 		key := string(aBytes)
 
diff --git a/p2p/host/resource-manager/error.go b/p2p/host/resource-manager/error.go
index 4b02672b..1e87e00a 100644
--- a/p2p/host/resource-manager/error.go
+++ b/p2p/host/resource-manager/error.go
@@ -6,13 +6,13 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 )
 
-type errStreamOrConnLimitExceeded struct {
+type ErrStreamOrConnLimitExceeded struct {
 	current, attempted, limit int
 	err                       error
 }
 
-func (e *errStreamOrConnLimitExceeded) Error() string { return e.err.Error() }
-func (e *errStreamOrConnLimitExceeded) Unwrap() error { return e.err }
+func (e *ErrStreamOrConnLimitExceeded) Error() string { return e.err.Error() }
+func (e *ErrStreamOrConnLimitExceeded) Unwrap() error { return e.err }
 
 // edge may be "" if this is not an edge error
 func logValuesStreamLimit(scope, edge string, dir network.Direction, stat network.ScopeStat, err error) []interface{} {
@@ -22,7 +22,7 @@ func logValuesStreamLimit(scope, edge string, dir network.Direction, stat networ
 		logValues = append(logValues, "edge", edge)
 	}
 	logValues = append(logValues, "direction", dir)
-	var e *errStreamOrConnLimitExceeded
+	var e *ErrStreamOrConnLimitExceeded
 	if errors.As(err, &e) {
 		logValues = append(logValues,
 			"current", e.current,
@@ -41,7 +41,7 @@ func logValuesConnLimit(scope, edge string, dir network.Direction, usefd bool, s
 		logValues = append(logValues, "edge", edge)
 	}
 	logValues = append(logValues, "direction", dir, "usefd", usefd)
-	var e *errStreamOrConnLimitExceeded
+	var e *ErrStreamOrConnLimitExceeded
 	if errors.As(err, &e) {
 		logValues = append(logValues,
 			"current", e.current,
@@ -52,14 +52,14 @@ func logValuesConnLimit(scope, edge string, dir network.Direction, usefd bool, s
 	return append(logValues, "stat", stat, "error", err)
 }
 
-type errMemoryLimitExceeded struct {
+type ErrMemoryLimitExceeded struct {
 	current, attempted, limit int64
 	priority                  uint8
 	err                       error
 }
 
-func (e *errMemoryLimitExceeded) Error() string { return e.err.Error() }
-func (e *errMemoryLimitExceeded) Unwrap() error { return e.err }
+func (e *ErrMemoryLimitExceeded) Error() string { return e.err.Error() }
+func (e *ErrMemoryLimitExceeded) Unwrap() error { return e.err }
 
 // edge may be "" if this is not an edge error
 func logValuesMemoryLimit(scope, edge string, stat network.ScopeStat, err error) []interface{} {
@@ -68,7 +68,7 @@ func logValuesMemoryLimit(scope, edge string, stat network.ScopeStat, err error)
 	if edge != "" {
 		logValues = append(logValues, "edge", edge)
 	}
-	var e *errMemoryLimitExceeded
+	var e *ErrMemoryLimitExceeded
 	if errors.As(err, &e) {
 		logValues = append(logValues,
 			"current", e.current,
diff --git a/p2p/host/resource-manager/scope.go b/p2p/host/resource-manager/scope.go
index 2a83c695..60089c3a 100644
--- a/p2p/host/resource-manager/scope.go
+++ b/p2p/host/resource-manager/scope.go
@@ -130,7 +130,7 @@ func (rc *resources) checkMemory(rsvp int64, prio uint8) error {
 	}
 
 	if !addOk || newmem > threshold {
-		return &errMemoryLimitExceeded{
+		return &ErrMemoryLimitExceeded{
 			current:   rc.memory,
 			attempted: rsvp,
 			limit:     limit,
@@ -171,7 +171,7 @@ func (rc *resources) addStreams(incount, outcount int) error {
 	if incount > 0 {
 		limit := rc.limit.GetStreamLimit(network.DirInbound)
 		if rc.nstreamsIn+incount > limit {
-			return &errStreamOrConnLimitExceeded{
+			return &ErrStreamOrConnLimitExceeded{
 				current:   rc.nstreamsIn,
 				attempted: incount,
 				limit:     limit,
@@ -182,7 +182,7 @@ func (rc *resources) addStreams(incount, outcount int) error {
 	if outcount > 0 {
 		limit := rc.limit.GetStreamLimit(network.DirOutbound)
 		if rc.nstreamsOut+outcount > limit {
-			return &errStreamOrConnLimitExceeded{
+			return &ErrStreamOrConnLimitExceeded{
 				current:   rc.nstreamsOut,
 				attempted: outcount,
 				limit:     limit,
@@ -192,7 +192,7 @@ func (rc *resources) addStreams(incount, outcount int) error {
 	}
 
 	if limit := rc.limit.GetStreamTotalLimit(); rc.nstreamsIn+incount+rc.nstreamsOut+outcount > limit {
-		return &errStreamOrConnLimitExceeded{
+		return &ErrStreamOrConnLimitExceeded{
 			current:   rc.nstreamsIn + rc.nstreamsOut,
 			attempted: incount + outcount,
 			limit:     limit,
@@ -244,7 +244,7 @@ func (rc *resources) addConns(incount, outcount, fdcount int) error {
 	if incount > 0 {
 		limit := rc.limit.GetConnLimit(network.DirInbound)
 		if rc.nconnsIn+incount > limit {
-			return &errStreamOrConnLimitExceeded{
+			return &ErrStreamOrConnLimitExceeded{
 				current:   rc.nconnsIn,
 				attempted: incount,
 				limit:     limit,
@@ -255,7 +255,7 @@ func (rc *resources) addConns(incount, outcount, fdcount int) error {
 	if outcount > 0 {
 		limit := rc.limit.GetConnLimit(network.DirOutbound)
 		if rc.nconnsOut+outcount > limit {
-			return &errStreamOrConnLimitExceeded{
+			return &ErrStreamOrConnLimitExceeded{
 				current:   rc.nconnsOut,
 				attempted: outcount,
 				limit:     limit,
@@ -265,7 +265,7 @@ func (rc *resources) addConns(incount, outcount, fdcount int) error {
 	}
 
 	if connLimit := rc.limit.GetConnTotalLimit(); rc.nconnsIn+incount+rc.nconnsOut+outcount > connLimit {
-		return &errStreamOrConnLimitExceeded{
+		return &ErrStreamOrConnLimitExceeded{
 			current:   rc.nconnsIn + rc.nconnsOut,
 			attempted: incount + outcount,
 			limit:     connLimit,
@@ -275,7 +275,7 @@ func (rc *resources) addConns(incount, outcount, fdcount int) error {
 	if fdcount > 0 {
 		limit := rc.limit.GetFDLimit()
 		if rc.nfd+fdcount > limit {
-			return &errStreamOrConnLimitExceeded{
+			return &ErrStreamOrConnLimitExceeded{
 				current:   rc.nfd,
 				attempted: fdcount,
 				limit:     limit,


Test patch:
diff --git a/p2p/host/peerstore/test/addr_book_suite.go b/p2p/host/peerstore/test/addr_book_suite.go
index 62327e31..27a74d96 100644
--- a/p2p/host/peerstore/test/addr_book_suite.go
+++ b/p2p/host/peerstore/test/addr_book_suite.go
@@ -141,6 +141,24 @@ func testAddAddress(ab pstore.AddrBook, clk *mockClock.Mock) func(*testing.T) {
 			ab.AddAddrs("", addrs, time.Hour)
 			AssertAddressesEqual(t, addrs, ab.Addrs(""))
 		})
+
+		t.Run("add a /p2p address with valid peerid", func(t *testing.T) {
+			peerId := GeneratePeerIDs(1)[0]
+			addr := GenerateAddrs(1)
+			p2pAddr := addr[0].Encapsulate(Multiaddr("/p2p/" + peerId.String()))
+			ab.AddAddr(peerId, p2pAddr, time.Hour)
+			AssertAddressesEqual(t, addr, ab.Addrs(peerId))
+		})
+
+		t.Run("add a /p2p address with invalid peerid", func(t *testing.T) {
+			pids := GeneratePeerIDs(2)
+			pid1 := pids[0]
+			pid2 := pids[1]
+			addr := GenerateAddrs(1)
+			p2pAddr := addr[0].Encapsulate(Multiaddr("/p2p/" + pid1.String()))
+			ab.AddAddr(pid2, p2pAddr, time.Hour)
+			AssertAddressesEqual(t, nil, ab.Addrs(pid2))
+		})
 	}
 }
 
diff --git a/test-plans/cmd/ping/main.go b/test-plans/cmd/ping/main.go
index acedee4f..d0432421 100644
--- a/test-plans/cmd/ping/main.go
+++ b/test-plans/cmd/ping/main.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"strconv"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -24,24 +25,33 @@ import (
 
 func main() {
 	var (
-		transport     = os.Getenv("transport")
-		secureChannel = os.Getenv("security")
-		muxer         = os.Getenv("muxer")
-		isDialerStr   = os.Getenv("is_dialer")
-		ip            = os.Getenv("ip")
-		redisAddr     = os.Getenv("REDIS_ADDR")
+		transport      = os.Getenv("transport")
+		secureChannel  = os.Getenv("security")
+		muxer          = os.Getenv("muxer")
+		isDialerStr    = os.Getenv("is_dialer")
+		ip             = os.Getenv("ip")
+		testTimeoutStr = os.Getenv("test_timeout")
+		redisAddr      = os.Getenv("REDIS_ADDR")
 	)
 
+	testTimeout := 10 * time.Second
+	if testTimeoutStr != "" {
+		secs, err := strconv.ParseInt(testTimeoutStr, 10, 32)
+		if err == nil {
+			testTimeout = time.Duration(secs) * time.Second
+		}
+	}
+
 	if redisAddr == "" {
 		redisAddr = "redis:6379"
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
 	defer cancel()
 
 	// Get peer information via redis
 	rClient := redis.NewClient(&redis.Options{
-		DialTimeout: 10 * time.Second,
+		DialTimeout: testTimeout,
 		Addr:        redisAddr,
 		Password:    "",
 		DB:          0,
@@ -110,7 +120,7 @@ func main() {
 	fmt.Println("My multiaddr is: ", host.Addrs())
 
 	if isDialer {
-		val, err := rClient.BLPop(ctx, 20*time.Second, "listenerAddr").Result()
+		val, err := rClient.BLPop(ctx, testTimeout, "listenerAddr").Result()
 		if err != nil {
 			log.Fatal("Failed to wait for listener to be ready")
 		}
@@ -144,7 +154,7 @@ func main() {
 		if err != nil {
 			log.Fatal("Failed to send listener address")
 		}
-		_, err = rClient.BLPop(ctx, 20*time.Second, "dialerDone").Result()
+		_, err = rClient.BLPop(ctx, testTimeout, "dialerDone").Result()
 		if err != nil {
 			log.Fatal("Failed to wait for dialer conclusion")
 		}