fix(floodsub): use monotonic sequence numbers (#3672)

* fix(floodsub): use monotonic sequence numbers

Seed the sequence number from the wall clock in nanoseconds and increment it, as the pubsub spec requires, instead of picking random values.

* test(floodsub): compare bigint sequence numbers directly

Author: tabcat

packages/floodsub/src/floodsub.ts

@@ -10,7 +10,7 @@ import { protocol, StrictNoSign, TopicValidatorResult, StrictSign } from './inde
 import { RPC } from './message/rpc.ts'
 import { PeerStreams } from './peer-streams.ts'
 import { signMessage, verifySignature } from './sign.ts'
-import { toMessage, noSignMsgId, msgId, toRpcMessage, randomSeqno } from './utils.ts'
+import { toMessage, noSignMsgId, msgId, toRpcMessage, nextSeqno } from './utils.ts'
 import type { RPCDecodeLimits } from './decodeRpc.ts'
 import type { FloodSubComponents, FloodSubEvents, FloodSubInit, FloodSub as FloodSubInterface, Message, PublishResult, SubscriptionChangeData, TopicValidatorFn } from './index.ts'
 import type { Logger, Connection, PeerId, Stream } from '@libp2p/interface'
@@ -601,7 +601,7 @@ export class FloodSub extends TypedEventEmitter<FloodSubEvents> implements Flood
       from: this.components.peerId,
       topic,
       data: data ?? new Uint8Array(0),
-      sequenceNumber: randomSeqno()
+      sequenceNumber: nextSeqno()
     }
 
     this.log('publish topic: %s from: %p data: %m', topic, message.from, message.data)

packages/floodsub/src/utils.ts

@@ -9,11 +9,15 @@ import type { PubSubRPCMessage } from './floodsub.ts'
 import type { Message } from './index.ts'
 import type { PublicKey } from '@libp2p/interface'
 
+// monotonic sequence number seeded from the wall clock in nanoseconds, the
+// pubsub spec requires a linearly increasing 64-bit big-endian uint
+let seqnoCounter = BigInt(Date.now()) * 1_000_000n
+
 /**
- * Generate a random sequence number
+ * Generate the next sequence number
  */
-export function randomSeqno (): bigint {
-  return BigInt(`0x${uint8ArrayToString(crypto.getRandomValues(new Uint8Array(8)), 'base16')}`)
+export function nextSeqno (): bigint {
+  return ++seqnoCounter
 }
 
 /**

packages/floodsub/test/forgery.spec.ts

@@ -11,7 +11,7 @@ import { floodsub } from '../src/index.ts'
 import { RPC } from '../src/message/rpc.ts'
 import { PeerStreams } from '../src/peer-streams.ts'
 import { SignPrefix } from '../src/sign.ts'
-import { randomSeqno, toRpcMessage } from '../src/utils.ts'
+import { nextSeqno, toRpcMessage } from '../src/utils.ts'
 import type { PubSubRPCMessage } from '../src/floodsub.ts'
 import type { FloodSub } from '../src/index.ts'
 import type { PeerId, PrivateKey, PublicKey } from '@libp2p/interface'
@@ -27,7 +27,7 @@ async function wireMessage (from: PeerId, signer: PrivateKey, key: PublicKey): P
     type: 'signed',
     from,
     data,
-    sequenceNumber: randomSeqno(),
+    sequenceNumber: nextSeqno(),
     topic,
     signature: new Uint8Array(0),
     key

packages/floodsub/test/message.spec.ts

@@ -10,7 +10,7 @@ import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { floodsub } from '../src/index.ts'
 import { RPC } from '../src/message/rpc.ts'
 import { SignPrefix } from '../src/sign.ts'
-import { randomSeqno, toMessage, toRpcMessage } from '../src/utils.ts'
+import { nextSeqno, toMessage, toRpcMessage } from '../src/utils.ts'
 import type { FloodSub, SignedMessage } from '../src/index.ts'
 import type { PeerId } from '@libp2p/interface'
 import type { Registrar } from '@libp2p/interface-internal'
@@ -43,7 +43,7 @@ describe('pubsub base messages', () => {
       from: peerId,
       data: uint8ArrayFromString('hello'),
       topic: 'test-topic',
-      sequenceNumber: randomSeqno()
+      sequenceNumber: nextSeqno()
     }
 
     // @ts-expect-error private method
@@ -58,7 +58,7 @@ describe('pubsub base messages', () => {
       from: peerId,
       data: uint8ArrayFromString('hello'),
       topic: 'test-topic',
-      sequenceNumber: randomSeqno()
+      sequenceNumber: nextSeqno()
     }
 
     sinon.stub(pubsub, 'globalSignaturePolicy').value('StrictSign')
@@ -92,7 +92,7 @@ describe('pubsub base messages', () => {
       from: peerId,
       data: uint8ArrayFromString('hello'),
       topic: 'test-topic',
-      sequenceNumber: randomSeqno()
+      sequenceNumber: nextSeqno()
     }
 
     sinon.stub(pubsub, 'globalSignaturePolicy').value('StrictNoSign')
@@ -127,7 +127,7 @@ describe('pubsub base messages', () => {
       from: victim,
       data: uint8ArrayFromString('hello'),
       topic: 'test-topic',
-      sequenceNumber: randomSeqno()
+      sequenceNumber: nextSeqno()
     }
     const bytes = uint8ArrayConcat([SignPrefix, RPC.Message.encode(toRpcMessage(forged)).subarray()])
     forged.signature = await attackerKey.sign(bytes)

packages/floodsub/test/sign.spec.ts

@@ -9,7 +9,7 @@ import {
   SignPrefix,
   verifySignature
 } from '../src/sign.ts'
-import { randomSeqno, toRpcMessage } from '../src/utils.ts'
+import { nextSeqno, toRpcMessage } from '../src/utils.ts'
 import type { PubSubRPCMessage } from '../src/floodsub.ts'
 import type { Message, SignedMessage } from '../src/index.ts'
 import type { PeerId, PrivateKey } from '@libp2p/interface'
@@ -33,7 +33,7 @@ describe('message signing', () => {
       type: 'signed',
       from: peerId,
       data: uint8ArrayFromString('hello'),
-      sequenceNumber: randomSeqno(),
+      sequenceNumber: nextSeqno(),
       topic: 'test-topic'
     }
 
@@ -65,7 +65,7 @@ describe('message signing', () => {
       type: 'signed',
       from: secPeerId,
       data: uint8ArrayFromString('hello'),
-      sequenceNumber: randomSeqno(),
+      sequenceNumber: nextSeqno(),
       topic: 'test-topic'
     }
 
@@ -93,7 +93,7 @@ describe('message signing', () => {
       type: 'signed',
       from: peerId,
       data: uint8ArrayFromString('hello'),
-      sequenceNumber: randomSeqno(),
+      sequenceNumber: nextSeqno(),
       topic: 'test-topic'
     }
 
@@ -127,7 +127,7 @@ describe('author key binding', () => {
       type: 'signed',
       from: victim,
       data,
-      sequenceNumber: randomSeqno(),
+      sequenceNumber: nextSeqno(),
       topic
     }
     const bytes = uint8ArrayConcat([SignPrefix, encodeMessage(toRpcMessage(message)).subarray()])
@@ -163,7 +163,7 @@ describe('author key binding', () => {
     })
 
     it('accepts a message whose key derives to the author', async () => {
-      const signed = await signMessage(rsaKey, { from: author, topic, data, sequenceNumber: randomSeqno() }, encodeMessage)
+      const signed = await signMessage(rsaKey, { from: author, topic, data, sequenceNumber: nextSeqno() }, encodeMessage)
       // signMessage rebuilds `from` from the private key, so it carries the key;
       // use the wire form so the derive-to-author check is exercised
       signed.from = author

packages/floodsub/test/utils.spec.ts

@@ -6,13 +6,17 @@ import type { PubSubRPCMessage } from '../src/floodsub.ts'
 import type { Message } from '../src/index.ts'
 
 describe('utils', () => {
-  it('randomSeqno', () => {
-    const first = utils.randomSeqno()
-    const second = utils.randomSeqno()
+  it('nextSeqno', () => {
+    const first = utils.nextSeqno()
+    const second = utils.nextSeqno()
 
     expect(first).to.be.a('BigInt')
     expect(second).to.be.a('BigInt')
-    expect(first).to.not.equal(second)
+    expect(second > first).to.be.true()
+  })
+
+  it('encodes sequence numbers as 8 bytes', () => {
+    expect(utils.bigIntToBytes(utils.nextSeqno())).to.have.lengthOf(8)
   })
 
   it('msgId should not generate same ID for two different Uint8Arrays', async () => {