feat: select early muxer (#482)

Implements early multiplexer negotiation to save multistream-select
round trips when a connection is opened.

Spec: https://github.com/libp2p/specs/blob/6d38f88f7b2d16b0e4489298bcd0737a6d704f7e/connections/inlined-muxer-negotiation.md#multiplexer-negotiation-over-noise

Author: achingbrain

package-lock.json

@@ -163,7 +163,7 @@
     "@noble/ciphers": "^1.1.3",
     "@noble/curves": "^1.1.0",
     "@noble/hashes": "^1.3.1",
-    "it-length-prefixed": "^9.0.1",
+    "it-length-prefixed": "^10.0.1",
     "it-length-prefixed-stream": "^1.0.0",
     "it-pair": "^2.0.6",
     "it-pipe": "^3.0.1",
@@ -194,7 +194,8 @@
     "multiformats": "^13.2.2",
     "p-defer": "^4.0.0",
     "protons": "^7.6.0",
-    "sinon": "^19.0.2"
+    "sinon": "^19.0.2",
+    "sinon-ts": "^2.0.0"
   },
   "browser": {
     "./dist/src/crypto/index.js": "./dist/src/crypto/index.browser.js"

package.json

@@ -37,16 +37,16 @@
  */
 
 import { Noise } from './noise.js'
-import type { NoiseInit } from './noise.js'
-import type { NoiseExtensions } from './proto/payload.js'
-import type { ComponentLogger, ConnectionEncrypter, Metrics, PeerId, PrivateKey } from '@libp2p/interface'
+import type { NoiseInit, NoiseExtensions } from './noise.js'
+import type { ComponentLogger, ConnectionEncrypter, Metrics, PeerId, PrivateKey, Upgrader } from '@libp2p/interface'
 export type { ICryptoInterface } from './crypto.js'
 export { pureJsCrypto } from './crypto/js.js'
 
 export interface NoiseComponents {
   peerId: PeerId
   privateKey: PrivateKey
   logger: ComponentLogger
+  upgrader: Upgrader
   metrics?: Metrics
 }
 

src/index.ts

@@ -1,5 +1,5 @@
 import { publicKeyFromProtobuf } from '@libp2p/crypto/keys'
-import { serviceCapabilities } from '@libp2p/interface'
+import { InvalidCryptoExchangeError, serviceCapabilities } from '@libp2p/interface'
 import { peerIdFromPublicKey } from '@libp2p/peer-id'
 import { decode } from 'it-length-prefixed'
 import { lpStream, type LengthPrefixedStream } from 'it-length-prefixed-stream'
@@ -14,18 +14,21 @@ import { type MetricsRegistry, registerMetrics } from './metrics.js'
 import { performHandshakeInitiator, performHandshakeResponder } from './performHandshake.js'
 import { decryptStream, encryptStream } from './streaming.js'
 import type { NoiseComponents } from './index.js'
-import type { NoiseExtensions } from './proto/payload.js'
 import type { HandshakeResult, ICrypto, INoiseConnection, KeyPair } from './types.js'
-import type { MultiaddrConnection, SecuredConnection, PeerId, PrivateKey, PublicKey, AbortOptions } from '@libp2p/interface'
+import type { MultiaddrConnection, SecuredConnection, PeerId, PrivateKey, PublicKey, AbortOptions, StreamMuxerFactory } from '@libp2p/interface'
 import type { Duplex } from 'it-stream-types'
 import type { Uint8ArrayList } from 'uint8arraylist'
 
+export interface NoiseExtensions {
+  webtransportCerthashes: Uint8Array[]
+}
+
 export interface NoiseInit {
   /**
    * x25519 private key, reuse for faster handshakes
    */
   staticNoiseKey?: Uint8Array
-  extensions?: NoiseExtensions
+  extensions?: Partial<NoiseExtensions>
   crypto?: ICryptoInterface
   prologueBytes?: Uint8Array
 }
@@ -47,7 +50,10 @@ export class Noise implements INoiseConnection {
     this.components = components
     const _crypto = crypto ?? defaultCrypto
     this.crypto = wrapCrypto(_crypto)
-    this.extensions = extensions
+    this.extensions = {
+      webtransportCerthashes: [],
+      ...extensions
+    }
     this.metrics = metrics ? registerMetrics(metrics) : undefined
 
     if (staticNoiseKey) {
@@ -100,7 +106,30 @@ export class Noise implements INoiseConnection {
     return {
       conn: connection,
       remoteExtensions: handshake.payload.extensions,
-      remotePeer: peerIdFromPublicKey(publicKey)
+      remotePeer: peerIdFromPublicKey(publicKey),
+      streamMuxer: this.getStreamMuxer(handshake.payload.extensions?.streamMuxers)
+    }
+  }
+
+  private getStreamMuxer (protocols?: string[]): StreamMuxerFactory | undefined {
+    if (protocols == null) {
+      return
+    }
+
+    const streamMuxers = this.components.upgrader.getStreamMuxers()
+
+    if (streamMuxers != null) {
+      for (const protocol of protocols) {
+        const streamMuxer = streamMuxers.get(protocol)
+
+        if (streamMuxer != null) {
+          return streamMuxer
+        }
+      }
+    }
+
+    if (protocols.length) {
+      throw new InvalidCryptoExchangeError('Early muxer negotiation was requested but the initiator and responder had no common muxers')
     }
   }
 
@@ -138,7 +167,8 @@ export class Noise implements INoiseConnection {
     return {
       conn: connection,
       remoteExtensions: handshake.payload.extensions,
-      remotePeer: peerIdFromPublicKey(publicKey)
+      remotePeer: peerIdFromPublicKey(publicKey),
+      streamMuxer: this.getStreamMuxer(handshake.payload.extensions?.streamMuxers)
     }
   }
 
@@ -162,7 +192,11 @@ export class Noise implements INoiseConnection {
         crypto: this.crypto,
         prologue: this.prologue,
         s: this.staticKey,
-        extensions: this.extensions
+        extensions: {
+          streamMuxers: [...this.components.upgrader.getStreamMuxers().keys()],
+          webtransportCerthashes: [],
+          ...this.extensions
+        }
       }, options)
       this.metrics?.xxHandshakeSuccesses.increment()
     } catch (e: unknown) {
@@ -192,7 +226,11 @@ export class Noise implements INoiseConnection {
         crypto: this.crypto,
         prologue: this.prologue,
         s: this.staticKey,
-        extensions: this.extensions
+        extensions: {
+          streamMuxers: [...this.components.upgrader.getStreamMuxers().keys()],
+          webtransportCerthashes: [],
+          ...this.extensions
+        }
       }, options)
       this.metrics?.xxHandshakeSuccesses.increment()
     } catch (e: unknown) {

src/noise.ts

@@ -2,10 +2,11 @@ syntax = "proto3";
 
 message NoiseExtensions {
     repeated bytes webtransport_certhashes = 1;
+    repeated string stream_muxers = 2;
 }
 
 message NoiseHandshakePayload {
   bytes identity_key = 1;
   bytes identity_sig = 2;
   optional NoiseExtensions extensions = 4;
-}
+}

src/proto/payload.proto

@@ -4,12 +4,13 @@
 /* eslint-disable @typescript-eslint/no-unnecessary-boolean-literal-compare */
 /* eslint-disable @typescript-eslint/no-empty-interface */
 
-import { type Codec, decodeMessage, encodeMessage, message } from 'protons-runtime'
+import { type Codec, decodeMessage, type DecodeOptions, encodeMessage, MaxLengthError, message } from 'protons-runtime'
 import { alloc as uint8ArrayAlloc } from 'uint8arrays/alloc'
 import type { Uint8ArrayList } from 'uint8arraylist'
 
 export interface NoiseExtensions {
   webtransportCerthashes: Uint8Array[]
+  streamMuxers: string[]
 }
 
 export namespace NoiseExtensions {
@@ -29,12 +30,20 @@ export namespace NoiseExtensions {
           }
         }
 
+        if (obj.streamMuxers != null) {
+          for (const value of obj.streamMuxers) {
+            w.uint32(18)
+            w.string(value)
+          }
+        }
+
         if (opts.lengthDelimited !== false) {
           w.ldelim()
         }
-      }, (reader, length) => {
+      }, (reader, length, opts = {}) => {
         const obj: any = {
-          webtransportCerthashes: []
+          webtransportCerthashes: [],
+          streamMuxers: []
         }
 
         const end = length == null ? reader.len : reader.pos + length
@@ -44,9 +53,21 @@ export namespace NoiseExtensions {
 
           switch (tag >>> 3) {
             case 1: {
+              if (opts.limits?.webtransportCerthashes != null && obj.webtransportCerthashes.length === opts.limits.webtransportCerthashes) {
+                throw new MaxLengthError('Decode error - map field "webtransportCerthashes" had too many elements')
+              }
+
               obj.webtransportCerthashes.push(reader.bytes())
               break
             }
+            case 2: {
+              if (opts.limits?.streamMuxers != null && obj.streamMuxers.length === opts.limits.streamMuxers) {
+                throw new MaxLengthError('Decode error - map field "streamMuxers" had too many elements')
+              }
+
+              obj.streamMuxers.push(reader.string())
+              break
+            }
             default: {
               reader.skipType(tag & 7)
               break
@@ -65,8 +86,8 @@ export namespace NoiseExtensions {
     return encodeMessage(obj, NoiseExtensions.codec())
   }
 
-  export const decode = (buf: Uint8Array | Uint8ArrayList): NoiseExtensions => {
-    return decodeMessage(buf, NoiseExtensions.codec())
+  export const decode = (buf: Uint8Array | Uint8ArrayList, opts?: DecodeOptions<NoiseExtensions>): NoiseExtensions => {
+    return decodeMessage(buf, NoiseExtensions.codec(), opts)
   }
 }
 
@@ -104,7 +125,7 @@ export namespace NoiseHandshakePayload {
         if (opts.lengthDelimited !== false) {
           w.ldelim()
         }
-      }, (reader, length) => {
+      }, (reader, length, opts = {}) => {
         const obj: any = {
           identityKey: uint8ArrayAlloc(0),
           identitySig: uint8ArrayAlloc(0)
@@ -125,7 +146,9 @@ export namespace NoiseHandshakePayload {
               break
             }
             case 4: {
-              obj.extensions = NoiseExtensions.codec().decode(reader, reader.uint32())
+              obj.extensions = NoiseExtensions.codec().decode(reader, reader.uint32(), {
+                limits: opts.limits?.extensions
+              })
               break
             }
             default: {
@@ -146,7 +169,7 @@ export namespace NoiseHandshakePayload {
     return encodeMessage(obj, NoiseHandshakePayload.codec())
   }
 
-  export const decode = (buf: Uint8Array | Uint8ArrayList): NoiseHandshakePayload => {
-    return decodeMessage(buf, NoiseHandshakePayload.codec())
+  export const decode = (buf: Uint8Array | Uint8ArrayList, opts?: DecodeOptions<NoiseHandshakePayload>): NoiseHandshakePayload => {
+    return decodeMessage(buf, NoiseHandshakePayload.codec(), opts)
   }
 }

src/proto/payload.ts

@@ -81,4 +81,8 @@ export interface KeyPair {
   privateKey: Uint8Array
 }
 
-export interface INoiseConnection extends ConnectionEncrypter<NoiseExtensions> { }
+export interface INoiseExtensions {
+  webtransportCerthashes: Uint8Array[]
+}
+
+export interface INoiseConnection extends ConnectionEncrypter<INoiseExtensions> { }

src/types.ts

@@ -2,7 +2,9 @@ import { generateKeyPair } from '@libp2p/crypto/keys'
 import tests from '@libp2p/interface-compliance-tests/connection-encryption'
 import { defaultLogger } from '@libp2p/logger'
 import { peerIdFromPrivateKey } from '@libp2p/peer-id'
+import { stubInterface } from 'sinon-ts'
 import { Noise } from '../src/noise.js'
+import type { Upgrader } from '@libp2p/interface'
 
 describe('spec compliance tests', function () {
   tests({
@@ -13,7 +15,10 @@ describe('spec compliance tests', function () {
       return new Noise({
         privateKey,
         peerId,
-        logger: defaultLogger()
+        logger: defaultLogger(),
+        upgrader: stubInterface<Upgrader>({
+          getStreamMuxers: () => new Map()
+        })
       })
     },
     async teardown () {}

test/compliance.spec.ts

@@ -5,10 +5,11 @@ import { expect } from 'aegir/chai'
 import { lpStream } from 'it-length-prefixed-stream'
 import { duplexPair } from 'it-pair/duplex'
 import sinon from 'sinon'
+import { stubInterface } from 'sinon-ts'
 import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
 import { noise } from '../src/index.js'
 import { Noise } from '../src/noise.js'
-import type { Metrics } from '@libp2p/interface'
+import type { Metrics, Upgrader } from '@libp2p/interface'
 import type { Uint8ArrayList } from 'uint8arraylist'
 
 function createCounterSpy (): ReturnType<typeof sinon.spy> {
@@ -26,7 +27,10 @@ describe('Index', () => {
     const noiseInstance = noise()({
       privateKey,
       peerId,
-      logger: defaultLogger()
+      logger: defaultLogger(),
+      upgrader: stubInterface<Upgrader>({
+        getStreamMuxers: () => new Map()
+      })
     })
     expect(noiseInstance.protocol).to.equal('/noise')
     expect(typeof (noiseInstance.secureInbound)).to.equal('function')
@@ -49,15 +53,21 @@ describe('Index', () => {
       privateKey: privateKeyInit,
       peerId: peerIdInit,
       logger: defaultLogger(),
-      metrics: metrics as any as Metrics
+      metrics: metrics as any as Metrics,
+      upgrader: stubInterface<Upgrader>({
+        getStreamMuxers: () => new Map()
+      })
     })
 
     const privateKeyResp = await generateKeyPair('Ed25519')
     const peerIdResp = peerIdFromPrivateKey(privateKeyResp)
     const noiseResp = new Noise({
       privateKey: privateKeyResp,
       peerId: peerIdResp,
-      logger: defaultLogger()
+      logger: defaultLogger(),
+      upgrader: stubInterface<Upgrader>({
+        getStreamMuxers: () => new Map()
+      })
     })
 
     const [inboundConnection, outboundConnection] = duplexPair<Uint8Array | Uint8ArrayList>()