Merge pull request #47 from libp2p/marco/allowlist

feat: Allowlist

rcmgr.go

@@ -9,6 +9,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
+	"github.com/multiformats/go-multiaddr"
 
 	logging "github.com/ipfs/go-log/v2"
 )
@@ -21,9 +22,14 @@ type resourceManager struct {
 	trace   *trace
 	metrics *metrics
 
+	allowlist *Allowlist
+
 	system    *systemScope
 	transient *transientScope
 
+	allowlistedSystem    *systemScope
+	allowlistedTransient *transientScope
+
 	cancelCtx context.Context
 	cancel    func()
 	wg        sync.WaitGroup
@@ -89,10 +95,12 @@ var _ network.PeerScope = (*peerScope)(nil)
 type connectionScope struct {
 	*resourceScope
 
-	dir   network.Direction
-	usefd bool
-	rcmgr *resourceManager
-	peer  *peerScope
+	dir           network.Direction
+	usefd         bool
+	isAllowlisted bool
+	rcmgr         *resourceManager
+	peer          *peerScope
+	endpoint      multiaddr.Multiaddr
 }
 
 var _ network.ConnScope = (*connectionScope)(nil)
@@ -117,11 +125,13 @@ var _ network.StreamManagementScope = (*streamScope)(nil)
 type Option func(*resourceManager) error
 
 func NewResourceManager(limits Limiter, opts ...Option) (network.ResourceManager, error) {
+	allowlist := newAllowlist()
 	r := &resourceManager{
-		limits: limits,
-		svc:    make(map[string]*serviceScope),
-		proto:  make(map[protocol.ID]*protocolScope),
-		peer:   make(map[peer.ID]*peerScope),
+		limits:    limits,
+		allowlist: &allowlist,
+		svc:       make(map[string]*serviceScope),
+		proto:     make(map[protocol.ID]*protocolScope),
+		peer:      make(map[peer.ID]*peerScope),
 	}
 
 	for _, opt := range opts {
@@ -134,11 +144,16 @@ func NewResourceManager(limits Limiter, opts ...Option) (network.ResourceManager
 		return nil, err
 	}
 
-	r.system = newSystemScope(limits.GetSystemLimits(), r)
+	r.system = newSystemScope(limits.GetSystemLimits(), r, "system")
 	r.system.IncRef()
-	r.transient = newTransientScope(limits.GetTransientLimits(), r)
+	r.transient = newTransientScope(limits.GetTransientLimits(), r, "transient", r.system.resourceScope)
 	r.transient.IncRef()
 
+	r.allowlistedSystem = newSystemScope(limits.GetSystemLimits(), r, "allowlistedSystem")
+	r.allowlistedSystem.IncRef()
+	r.allowlistedTransient = newTransientScope(limits.GetTransientLimits(), r, "allowlistedTransient", r.allowlistedSystem.resourceScope)
+	r.allowlistedTransient.IncRef()
+
 	r.cancelCtx, r.cancel = context.WithCancel(context.Background())
 
 	r.wg.Add(1)
@@ -147,6 +162,10 @@ func NewResourceManager(limits Limiter, opts ...Option) (network.ResourceManager
 	return r, nil
 }
 
+func (r *resourceManager) GetAllowlist() *Allowlist {
+	return r.allowlist
+}
+
 func (r *resourceManager) ViewSystem(f func(network.ResourceScope) error) error {
 	return f(r.system)
 }
@@ -255,10 +274,23 @@ func (r *resourceManager) nextStreamId() int64 {
 	return r.streamId
 }
 
-func (r *resourceManager) OpenConnection(dir network.Direction, usefd bool) (network.ConnManagementScope, error) {
-	conn := newConnectionScope(dir, usefd, r.limits.GetConnLimits(), r)
+func (r *resourceManager) OpenConnection(dir network.Direction, usefd bool, endpoint multiaddr.Multiaddr) (network.ConnManagementScope, error) {
+	var conn *connectionScope
+	conn = newConnectionScope(dir, usefd, r.limits.GetConnLimits(), r, endpoint)
 
-	if err := conn.AddConn(dir, usefd); err != nil {
+	err := conn.AddConn(dir, usefd)
+	if err != nil {
+		// Try again if this is an allowlisted connection
+		// Failed to open connection, let's see if this was allowlisted and try again
+		allowed := r.allowlist.Allowed(endpoint)
+		if allowed {
+			conn.Done()
+			conn = newAllowListedConnectionScope(dir, usefd, r.limits.GetConnLimits(), r, endpoint)
+			err = conn.AddConn(dir, usefd)
+		}
+	}
+
+	if err != nil {
 		conn.Done()
 		r.metrics.BlockConn(dir, usefd)
 		return nil, err
@@ -363,17 +395,17 @@ func (r *resourceManager) gc() {
 	}
 }
 
-func newSystemScope(limit Limit, rcmgr *resourceManager) *systemScope {
+func newSystemScope(limit Limit, rcmgr *resourceManager, name string) *systemScope {
 	return &systemScope{
-		resourceScope: newResourceScope(limit, nil, "system", rcmgr.trace, rcmgr.metrics),
+		resourceScope: newResourceScope(limit, nil, name, rcmgr.trace, rcmgr.metrics),
 	}
 }
 
-func newTransientScope(limit Limit, rcmgr *resourceManager) *transientScope {
+func newTransientScope(limit Limit, rcmgr *resourceManager, name string, systemScope *resourceScope) *transientScope {
 	return &transientScope{
 		resourceScope: newResourceScope(limit,
-			[]*resourceScope{rcmgr.system.resourceScope},
-			"transient", rcmgr.trace, rcmgr.metrics),
+			[]*resourceScope{systemScope},
+			name, rcmgr.trace, rcmgr.metrics),
 		system: rcmgr.system,
 	}
 }
@@ -408,14 +440,28 @@ func newPeerScope(p peer.ID, limit Limit, rcmgr *resourceManager) *peerScope {
 	}
 }
 
-func newConnectionScope(dir network.Direction, usefd bool, limit Limit, rcmgr *resourceManager) *connectionScope {
+func newConnectionScope(dir network.Direction, usefd bool, limit Limit, rcmgr *resourceManager, endpoint multiaddr.Multiaddr) *connectionScope {
 	return &connectionScope{
 		resourceScope: newResourceScope(limit,
 			[]*resourceScope{rcmgr.transient.resourceScope, rcmgr.system.resourceScope},
 			fmt.Sprintf("conn-%d", rcmgr.nextConnId()), rcmgr.trace, rcmgr.metrics),
-		dir:   dir,
-		usefd: usefd,
-		rcmgr: rcmgr,
+		dir:      dir,
+		usefd:    usefd,
+		rcmgr:    rcmgr,
+		endpoint: endpoint,
+	}
+}
+
+func newAllowListedConnectionScope(dir network.Direction, usefd bool, limit Limit, rcmgr *resourceManager, endpoint multiaddr.Multiaddr) *connectionScope {
+	return &connectionScope{
+		resourceScope: newResourceScope(limit,
+			[]*resourceScope{rcmgr.allowlistedTransient.resourceScope, rcmgr.allowlistedSystem.resourceScope},
+			fmt.Sprintf("conn-%d", rcmgr.nextConnId()), rcmgr.trace, rcmgr.metrics),
+		dir:           dir,
+		usefd:         usefd,
+		rcmgr:         rcmgr,
+		endpoint:      endpoint,
+		isAllowlisted: true,
 	}
 }
 
@@ -500,6 +546,49 @@ func (s *connectionScope) PeerScope() network.PeerScope {
 	return s.peer
 }
 
+// transferAllowedToStandard transfers this connection scope from being part of
+// the allowlist set of scopes to being part of the standard set of scopes.
+// Happens when we first allowlisted this connection due to its IP, but later
+// discovered that the peer id not what we expected.
+func (s *connectionScope) transferAllowedToStandard() (err error) {
+
+	systemScope := s.rcmgr.system.resourceScope
+	transientScope := s.rcmgr.transient.resourceScope
+
+	stat := s.resourceScope.rc.stat()
+
+	for _, scope := range s.edges {
+		scope.ReleaseForChild(stat)
+		scope.DecRef() // removed from edges
+	}
+	s.edges = nil
+
+	if err := systemScope.ReserveForChild(stat); err != nil {
+		return err
+	}
+	systemScope.IncRef()
+
+	// Undo this if we fail later
+	defer func() {
+		if err != nil {
+			systemScope.ReleaseForChild(stat)
+			systemScope.DecRef()
+		}
+	}()
+
+	if err := transientScope.ReserveForChild(stat); err != nil {
+		return err
+	}
+	transientScope.IncRef()
+
+	// Update edges
+	s.edges = []*resourceScope{
+		systemScope,
+		transientScope,
+	}
+	return nil
+}
+
 func (s *connectionScope) SetPeer(p peer.ID) error {
 	s.Lock()
 	defer s.Unlock()
@@ -507,6 +596,34 @@ func (s *connectionScope) SetPeer(p peer.ID) error {
 	if s.peer != nil {
 		return fmt.Errorf("connection scope already attached to a peer")
 	}
+
+	system := s.rcmgr.system
+	transient := s.rcmgr.transient
+
+	if s.isAllowlisted {
+		system = s.rcmgr.allowlistedSystem
+		transient = s.rcmgr.allowlistedTransient
+
+		if !s.rcmgr.allowlist.AllowedPeerAndMultiaddr(p, s.endpoint) {
+			s.isAllowlisted = false
+
+			// This is not an allowed peer + multiaddr combination. We need to
+			// transfer this connection to the general scope. We'll do this first by
+			// transferring the connection to the system and transient scopes, then
+			// continue on with this function. The idea is that a connection
+			// shouldn't get the benefit of evading the transient scope because it
+			// was _almost_ an allowlisted connection.
+			if err := s.transferAllowedToStandard(); err != nil {
+				// Failed to transfer this connection to the standard scopes
+				return err
+			}
+
+			// set the system and transient scopes to the non-allowlisted ones
+			system = s.rcmgr.system
+			transient = s.rcmgr.transient
+		}
+	}
+
 	s.peer = s.rcmgr.getPeerScope(p)
 
 	// juggle resources from transient scope to peer scope
@@ -518,13 +635,13 @@ func (s *connectionScope) SetPeer(p peer.ID) error {
 		return err
 	}
 
-	s.rcmgr.transient.ReleaseForChild(stat)
-	s.rcmgr.transient.DecRef() // removed from edges
+	transient.ReleaseForChild(stat)
+	transient.DecRef() // removed from edges
 
 	// update edges
 	edges := []*resourceScope{
 		s.peer.resourceScope,
-		s.rcmgr.system.resourceScope,
+		system.resourceScope,
 	}
 	s.resourceScope.edges = edges