Adds Application Specific RPC Inspector (libp2p#509)

* Update go.mod

* Refactor GossipSub Construction  (libp2p#1)

* Enables non-atomic validation for peer scoring parameters (libp2p#499)

* decouples topic scoring parameters

* adds skiping atomic validation for topic parameters

* cleans up

* adds skip atomic validation to peer score threshold

* adds skip atomic validation for peer parameters

* adds test for non-atomic validation

* adds tests for peer score

* adds tests for peer score thresholds

* refactors tests

* chore: Update .github/workflows/stale.yml [skip ci]

* adds with gossipsub tracker

Co-authored-by: libp2p-mgmt-read-write[bot] <104492852+libp2p-mgmt-read-write[bot]@users.noreply.github.com>

* decouples options

* fixes conflict

* reverts back module

* fixes peer score helper

* Adds send control message to gossipsub router (libp2p#2)

* adjusts libp2p version (libp2p#3)

* Update go.mod (libp2p#4)

* adds app specific rpc handler

* Create ci.yml (libp2p#5)

* Create Makefile (libp2p#7)

* Revert "Merge branch 'yahya/gossipsub-router-interface'" (libp2p#6)

This reverts commit 1c91995.

* Update ci.yml (libp2p#9)

* Revert "Merge branch 'master' into yahya/adds-rpc-inspector"

This reverts commit 352d747.

* Revert "Merge remote-tracking branch 'origin/yahya/adds-rpc-inspector' into yahya/adds-rpc-inspector"

This reverts commit 586c5cb.

* Revert "Merge branch 'master' into yahya/adds-rpc-inspector"

This reverts commit 2e13ee8.

* moves app specific inspector to pubsub

* removes option from gossipsub

* moves app specific rpc inspector up

* refactors app specific to return an error

Co-authored-by: libp2p-mgmt-read-write[bot] <104492852+libp2p-mgmt-read-write[bot]@users.noreply.github.com>

pubsub.go

@@ -170,6 +170,12 @@ type PubSub struct {
 	protoMatchFunc ProtocolMatchFn
 
 	ctx context.Context
+
+	// appSpecificRpcInspector is an auxiliary that may be set by the application to inspect incoming RPCs prior to
+	// processing them. The inspector is invoked on an accepted RPC right prior to handling it.
+	// The return value of the inspector function is an error indicating whether the RPC should be processed or not.
+	// If the error is nil, the RPC is processed as usual. If the error is non-nil, the RPC is dropped.
+	appSpecificRpcInspector func(peer.ID, *RPC) error
 }
 
 // PubSubRouter is the message router component of PubSub.
@@ -527,6 +533,13 @@ func WithSeenMessagesTTL(ttl time.Duration) Option {
 	}
 }
 
+func WithAppSpecificRpcInspector(inspector func(peer.ID, *RPC) error) Option {
+	return func(ps *PubSub) error {
+		ps.appSpecificRpcInspector = inspector
+		return nil
+	}
+}
+
 // processLoop handles all inputs arriving on the channels
 func (p *PubSub) processLoop(ctx context.Context) {
 	defer func() {
@@ -1005,6 +1018,15 @@ func (p *PubSub) notifyLeave(topic string, pid peer.ID) {
 }
 
 func (p *PubSub) handleIncomingRPC(rpc *RPC) {
+	// pass the rpc through app specific validation (if any available).
+	if p.appSpecificRpcInspector != nil {
+		// check if the RPC is allowed by the external inspector
+		if err := p.appSpecificRpcInspector(rpc.from, rpc); err != nil {
+			log.Debugf("application-specific inspection failed, rejecting incoming rpc: %s", err)
+			return // reject the RPC
+		}
+	}
+
 	p.tracer.RecvRPC(rpc)
 
 	subs := rpc.GetSubscriptions()