feat: add invalid message spam protection

test/peerScore.spec.js

@@ -4,6 +4,7 @@ const { utils } = require('libp2p-pubsub')
 const delay = require('delay')
 
 const { PeerScore, createPeerScoreParams, createTopicScoreParams } = require('../src/score')
+const { ERR_TOPIC_VALIDATOR_IGNORE, ERR_TOPIC_VALIDATOR_REJECT } = require('../src/constants')
 const { makeTestMessage } = require('./utils')
 
 const connectionManager = new Map()
@@ -412,7 +413,7 @@ describe('PeerScore', () => {
     for (let i = 0; i < nMessages; i++) {
       const msg = makeTestMessage(i, [mytopic])
       msg.receivedFrom = peerA
-      ps.rejectMessage(msg)
+      ps.rejectMessage(msg, ERR_TOPIC_VALIDATOR_REJECT)
     }
     ps._refreshScores()
     let aScore = ps.score(peerA)
@@ -441,7 +442,7 @@ describe('PeerScore', () => {
     for (let i = 0; i < nMessages; i++) {
       const msg = makeTestMessage(i, [mytopic])
       msg.receivedFrom = peerA
-      ps.rejectMessage(msg)
+      ps.rejectMessage(msg, ERR_TOPIC_VALIDATOR_REJECT)
     }
     ps._refreshScores()
     let aScore = ps.score(peerA)
@@ -481,7 +482,7 @@ describe('PeerScore', () => {
     ps.validateMessage(msg)
 
     // this should have no effect in the score, and subsequent duplicate messages should have no effect either
-    ps.ignoreMessage(msg)
+    ps.rejectMessage(msg, ERR_TOPIC_VALIDATOR_IGNORE)
     msg.receivedFrom = peerB
     ps.duplicateMessage(msg)
 
@@ -501,7 +502,7 @@ describe('PeerScore', () => {
     ps.validateMessage(msg)
 
     // and reject the message to make sure duplicates are also penalized
-    ps.rejectMessage(msg)
+    ps.rejectMessage(msg, ERR_TOPIC_VALIDATOR_REJECT)
     msg.receivedFrom = peerB
     ps.duplicateMessage(msg)
 
@@ -524,7 +525,7 @@ describe('PeerScore', () => {
     msg.receivedFrom = peerB
     ps.duplicateMessage(msg)
     msg.receivedFrom = peerA
-    ps.rejectMessage(msg)
+    ps.rejectMessage(msg, ERR_TOPIC_VALIDATOR_REJECT)
 
     aScore = ps.score(peerA)
     bScore = ps.score(peerB)

test/pubsub.spec.js

@@ -15,6 +15,7 @@ const { signMessage } = require('libp2p-pubsub/src/message/sign')
 const PeerId = require('peer-id')
 
 const Gossipsub = require('../src')
+const { ERR_TOPIC_VALIDATOR_REJECT } = require('../src/constants')
 const {
   createPeer,
   startNode,
@@ -146,7 +147,7 @@ describe('Pubsub', () => {
       // Set a trivial topic validator
       gossipsub.topicValidators.set(filteredTopic, (topic, message) => {
         if (!message.data.equals(Buffer.from('a message'))) {
-          throw errcode(new Error(), 'reject')
+          throw errcode(new Error(), ERR_TOPIC_VALIDATOR_REJECT)
         }
       })
 

ts/constants.ts

@@ -213,8 +213,5 @@ export const GossipsubIWantFollowupTime = 3 * second
 
 export const TimeCacheDuration = 120 * 1000
 
-export const enum ExtendedValidatorResult {
-  accept = 'accept',
-  reject = 'reject',
-  ignore = 'ignore'
-}
+export const ERR_TOPIC_VALIDATOR_REJECT = 'ERR_TOPIC_VALIDATOR_REJECT'
+export const ERR_TOPIC_VALIDATOR_IGNORE = 'ERR_TOPIC_VALIDATOR_IGNORE'

ts/index.ts

@@ -8,7 +8,6 @@ import {
   ControlMessage, ControlIHave, ControlGraft, ControlIWant, ControlPrune
 } from './message'
 import * as constants from './constants'
-import { ExtendedValidatorResult } from './constants'
 import { Heartbeat } from './heartbeat'
 import { getGossipPeers } from './getGossipPeers'
 import { createGossipRpc, shuffle, hasGossipProtocol } from './utils'
@@ -417,16 +416,8 @@ class Gossipsub extends BasicPubsub {
     try {
       await super.validate(message)
     } catch (e) {
-      switch (e.code) {
-        case ExtendedValidatorResult.reject:
-          this.score.rejectMessage(message)
-          this.gossipTracer.rejectMessage(message)
-          break
-        case ExtendedValidatorResult.ignore:
-          this.score.ignoreMessage(message)
-          this.gossipTracer.rejectMessage(message)
-          break
-      }
+      this.score.rejectMessage(message, e.code)
+      this.gossipTracer.rejectMessage(message, e.code)
       throw e
     }
   }

ts/score/peerScore.ts

@@ -4,10 +4,19 @@ import { PeerStats, createPeerStats, ensureTopicStats } from './peerStats'
 import { computeScore } from './computeScore'
 import { MessageDeliveries, DeliveryRecordStatus } from './messageDeliveries'
 import { ConnectionManager } from '../interfaces'
+import { ERR_TOPIC_VALIDATOR_IGNORE } from '../constants'
 import PeerId = require('peer-id')
 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 // @ts-ignore
 import debug = require('debug')
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-ignore
+import pubsubErrors = require('libp2p-pubsub/src/errors')
+
+const {
+  ERR_INVALID_SIGNATURE,
+  ERR_MISSING_SIGNATURE
+} = pubsubErrors.codes
 
 const log = debug('libp2p:gossipsub:score')
 
@@ -320,10 +329,18 @@ export class PeerScore {
 
   /**
    * @param {InMessage} message
+   * @param {string} reason
    * @returns {void}
    */
-  rejectMessage (message: InMessage): void {
+  rejectMessage (message: InMessage, reason: string): void {
     const id = message.receivedFrom
+    switch (reason) {
+      case ERR_MISSING_SIGNATURE:
+      case ERR_INVALID_SIGNATURE:
+        this._markInvalidMessageDelivery(id, message)
+        return
+    }
+
     const drec = this.deliveryRecords.ensureRecord(this.msgId(message))
 
     // defensive check that this is the first rejection -- delivery status should be unknown
@@ -335,6 +352,13 @@ export class PeerScore {
       return
     }
 
+    switch (reason) {
+      case ERR_TOPIC_VALIDATOR_IGNORE:
+        // we were explicitly instructed by the validator to ignore the message but not penalize the peer
+        drec.status = DeliveryRecordStatus.ignored
+        return
+    }
+
     // mark the message as invalid and penalize peers that have already forwarded it.
     drec.status = DeliveryRecordStatus.invalid
 
@@ -344,27 +368,6 @@ export class PeerScore {
     })
   }
 
-  /**
-   * @param {InMessage} message
-   * @returns {void}
-   */
-  ignoreMessage (message: InMessage): void {
-    const id = message.receivedFrom
-    const drec = this.deliveryRecords.ensureRecord(this.msgId(message))
-
-    // defensive check that this is the first ignore -- delivery status should be unknown
-    if (drec.status !== DeliveryRecordStatus.unknown) {
-      log(
-        'unexpected ignore: message from %s was first seen %s ago and has delivery status %d',
-        id, Date.now() - drec.firstSeen, DeliveryRecordStatus[drec.status]
-      )
-      return
-    }
-
-    // mark the message as invalid and penalize peers that have already forwarded it.
-    drec.status = DeliveryRecordStatus.ignored
-  }
-
   /**
    * @param {InMessage} message
    * @returns {void}

ts/tracer.ts

@@ -1,5 +1,13 @@
 import { InMessage } from './message'
 import { GossipsubIWantFollowupTime } from './constants'
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-ignore
+import pubsubErrors = require('libp2p-pubsub/src/errors')
+
+const {
+  ERR_INVALID_SIGNATURE,
+  ERR_MISSING_SIGNATURE
+} = pubsubErrors.codes
 
 /**
  * IWantTracer is an internal tracer that tracks IWANT requests in order to penalize
@@ -86,7 +94,13 @@ export class IWantTracer {
    * @param {InMessage} msg
    * @returns {void}
    */
-  rejectMessage (msg: InMessage): void {
+  rejectMessage (msg: InMessage, reason: string): void {
+    switch (reason) {
+      case ERR_INVALID_SIGNATURE:
+      case ERR_MISSING_SIGNATURE:
+        return
+    }
+
     const msgId = this.getMsgId(msg)
     this.promises.delete(msgId)
   }