feat: select muxer early (#3026)

Uses ALPN protocols to select an early muxer and skip the
multistream-select dance.

Restores compatibility with jvm-libp2p.

Author: achingbrain

interop/node-version.json

@@ -11,6 +11,6 @@
     "webrtc",
     "webrtc-direct"
   ],
-  "secureChannels": ["noise"],
+  "secureChannels": ["noise", "tls"],
   "muxers": ["yamux", "mplex"]
 }

interop/package.json

@@ -21,11 +21,12 @@
     "@chainsafe/libp2p-noise": "^16.0.0",
     "@chainsafe/libp2p-yamux": "^7.0.1",
     "@libp2p/circuit-relay-v2": "^3.1.3",
-    "@libp2p/interface": "^2.2.1",
     "@libp2p/identify": "^3.0.12",
+    "@libp2p/interface": "^2.2.1",
     "@libp2p/mplex": "^11.0.13",
     "@libp2p/ping": "^2.0.12",
     "@libp2p/tcp": "^10.0.13",
+    "@libp2p/tls": "^2.0.17",
     "@libp2p/webrtc": "^5.0.19",
     "@libp2p/websockets": "^9.0.13",
     "@libp2p/webtransport": "^5.0.18",

interop/test/fixtures/get-libp2p.ts

@@ -3,15 +3,19 @@
 import { noise } from '@chainsafe/libp2p-noise'
 import { yamux } from '@chainsafe/libp2p-yamux'
 import { circuitRelayTransport } from '@libp2p/circuit-relay-v2'
-import { type Identify, identify } from '@libp2p/identify'
+import { identify } from '@libp2p/identify'
 import { mplex } from '@libp2p/mplex'
-import { type PingService, ping } from '@libp2p/ping'
+import { ping } from '@libp2p/ping'
 import { tcp } from '@libp2p/tcp'
+import { tls } from '@libp2p/tls'
 import { webRTC, webRTCDirect } from '@libp2p/webrtc'
 import { webSockets } from '@libp2p/websockets'
 import { webTransport } from '@libp2p/webtransport'
-import { type Libp2pOptions, createLibp2p } from 'libp2p'
+import { createLibp2p } from 'libp2p'
+import type { Identify } from '@libp2p/identify'
 import type { Libp2p } from '@libp2p/interface'
+import type { PingService } from '@libp2p/ping'
+import type { Libp2pOptions } from 'libp2p'
 
 const isDialer: boolean = process.env.is_dialer === 'true'
 
@@ -106,6 +110,9 @@ export async function getLibp2p (): Promise<Libp2p<{ ping: PingService }>> {
       case 'noise':
         options.connectionEncrypters = [noise()]
         break
+      case 'tls':
+        options.connectionEncrypters = [tls()]
+        break
       default:
         throw new Error(`Unknown secure channel: ${SECURE_CHANNEL ?? ''}`)
     }

packages/connection-encrypter-tls/package.json

@@ -67,7 +67,8 @@
     "aegir": "^45.1.1",
     "it-pair": "^2.0.6",
     "protons": "^7.6.0",
-    "sinon": "^19.0.2"
+    "sinon": "^19.0.2",
+    "sinon-ts": "^2.0.0"
   },
   "browser": {
     "./dist/src/tls.js": "./dist/src/tls.browser.js"

packages/connection-encrypter-tls/src/index.ts

@@ -19,13 +19,14 @@
  */
 
 import { TLS } from './tls.js'
-import type { ComponentLogger, ConnectionEncrypter, Metrics, PrivateKey } from '@libp2p/interface'
+import type { ComponentLogger, ConnectionEncrypter, Metrics, PrivateKey, Upgrader } from '@libp2p/interface'
 
 export const PROTOCOL = '/tls/1.0.0'
 
 export interface TLSComponents {
   privateKey: PrivateKey
   logger: ComponentLogger
+  upgrader: Upgrader
   metrics?: Metrics
 }
 

packages/connection-encrypter-tls/src/tls.ts

@@ -19,12 +19,12 @@
  */
 
 import { TLSSocket, type TLSSocketOptions, connect } from 'node:tls'
-import { serviceCapabilities } from '@libp2p/interface'
+import { InvalidCryptoExchangeError, serviceCapabilities } from '@libp2p/interface'
 import { HandshakeTimeoutError } from './errors.js'
 import { generateCertificate, verifyPeerCertificate, itToStream, streamToIt } from './utils.js'
 import { PROTOCOL } from './index.js'
 import type { TLSComponents } from './index.js'
-import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, Logger, SecureConnectionOptions, CounterGroup } from '@libp2p/interface'
+import type { MultiaddrConnection, ConnectionEncrypter, SecuredConnection, Logger, SecureConnectionOptions, CounterGroup, StreamMuxerFactory } from '@libp2p/interface'
 import type { Duplex } from 'it-stream-types'
 import type { Uint8ArrayList } from 'uint8arraylist'
 
@@ -88,14 +88,41 @@ export class TLS implements ConnectionEncrypter {
    * Encrypt connection
    */
   async _encrypt <Stream extends Duplex<AsyncGenerator<Uint8Array | Uint8ArrayList>> = MultiaddrConnection> (conn: Stream, isServer: boolean, options?: SecureConnectionOptions): Promise<SecuredConnection<Stream>> {
+    let streamMuxer: StreamMuxerFactory | undefined
+
     const opts: TLSSocketOptions = {
       ...await generateCertificate(this.components.privateKey),
       isServer,
       // require TLS 1.3 or later
       minVersion: 'TLSv1.3',
       maxVersion: 'TLSv1.3',
       // accept self-signed certificates
-      rejectUnauthorized: false
+      rejectUnauthorized: false,
+
+      // early negotiation of muxer via ALPN protocols
+      ALPNProtocols: [
+        ...this.components.upgrader.getStreamMuxers().keys(),
+        'libp2p'
+      ],
+      ALPNCallback: ({ protocols }) => {
+        this.log.trace('received protocols %s', protocols)
+        let chosenProtocol: string | undefined
+
+        for (const protocol of protocols) {
+          if (protocol === 'libp2p') {
+            chosenProtocol = 'libp2p'
+          }
+
+          streamMuxer = this.components.upgrader.getStreamMuxers().get(protocol)
+
+          if (streamMuxer != null) {
+            chosenProtocol = protocol
+            break
+          }
+        }
+
+        return chosenProtocol
+      }
     }
 
     let socket: TLSSocket
@@ -131,12 +158,38 @@ export class TLS implements ConnectionEncrypter {
           .then(remotePeer => {
             this.log('remote certificate ok, remote peer %p', remotePeer)
 
+            if (!isServer && typeof socket.alpnProtocol === 'string') {
+              streamMuxer = this.components.upgrader.getStreamMuxers().get(socket.alpnProtocol)
+
+              if (streamMuxer == null) {
+                this.log.error('selected muxer that did not exist')
+              }
+            }
+
+            // 'libp2p' is a special protocol - if it's sent the remote does not
+            // support early muxer negotiation
+            if (!isServer && typeof socket.alpnProtocol === 'string' && socket.alpnProtocol !== 'libp2p') {
+              this.log.trace('got early muxer', socket.alpnProtocol)
+              streamMuxer = this.components.upgrader.getStreamMuxers().get(socket.alpnProtocol)
+
+              if (streamMuxer == null) {
+                const err = new InvalidCryptoExchangeError(`Selected muxer ${socket.alpnProtocol} did not exist`)
+                this.log.error(`Selected muxer ${socket.alpnProtocol} did not exist - %e`, err)
+
+                if (isAbortable(conn)) {
+                  conn.abort(err)
+                  reject(err)
+                }
+              }
+            }
+
             resolve({
               remotePeer,
               conn: {
                 ...conn,
                 ...streamToIt(socket)
-              }
+              },
+              streamMuxer
             })
           })
           .catch((err: Error) => {

packages/connection-encrypter-tls/test/index.spec.ts

@@ -6,8 +6,9 @@ import { peerIdFromMultihash, peerIdFromPrivateKey } from '@libp2p/peer-id'
 import { expect } from 'aegir/chai'
 import { duplexPair } from 'it-pair/duplex'
 import sinon from 'sinon'
+import { stubInterface } from 'sinon-ts'
 import { tls } from '../src/index.js'
-import type { ConnectionEncrypter, PeerId } from '@libp2p/interface'
+import type { StreamMuxerFactory, ConnectionEncrypter, PeerId, Upgrader, MultiaddrConnection } from '@libp2p/interface'
 
 describe('tls', () => {
   let localPeer: PeerId
@@ -26,7 +27,14 @@ describe('tls', () => {
 
     encrypter = tls()({
       privateKey: localKeyPair,
-      logger: defaultLogger()
+      logger: defaultLogger(),
+      upgrader: stubInterface<Upgrader>({
+        getStreamMuxers () {
+          return new Map([['/test/muxer', stubInterface<StreamMuxerFactory>({
+            protocol: '/test/muxer'
+          })]])
+        }
+      })
     })
   })
 
@@ -38,10 +46,14 @@ describe('tls', () => {
     const [inbound, outbound] = duplexPair<any>()
 
     await Promise.all([
-      encrypter.secureInbound(inbound, {
+      encrypter.secureInbound(stubInterface<MultiaddrConnection>({
+        ...inbound
+      }), {
         remotePeer
       }),
-      encrypter.secureOutbound(outbound, {
+      encrypter.secureOutbound(stubInterface<MultiaddrConnection>({
+        ...outbound
+      }), {
         remotePeer: wrongPeer
       })
     ]).then(() => expect.fail('should have failed'), (err) => {
@@ -57,19 +69,48 @@ describe('tls', () => {
 
     encrypter = tls()({
       privateKey: keyPair,
-      logger: defaultLogger()
+      logger: defaultLogger(),
+      upgrader: stubInterface<Upgrader>({
+        getStreamMuxers () {
+          return new Map([['/test/muxer', stubInterface<StreamMuxerFactory>()]])
+        }
+      })
     })
 
     const [inbound, outbound] = duplexPair<any>()
 
     await expect(Promise.all([
-      encrypter.secureInbound(inbound, {
+      encrypter.secureInbound(stubInterface<MultiaddrConnection>({
+        ...inbound
+      }), {
         remotePeer
       }),
-      encrypter.secureOutbound(outbound, {
+      encrypter.secureOutbound(stubInterface<MultiaddrConnection>({
+        ...outbound
+      }), {
         remotePeer: localPeer
       })
     ]))
       .to.eventually.be.rejected.with.property('name', 'UnexpectedPeerError')
   })
+
+  it('should select an early muxer', async () => {
+    const [inbound, outbound] = duplexPair<any>()
+
+    const result = await Promise.all([
+      encrypter.secureInbound(stubInterface<MultiaddrConnection>({
+        ...inbound
+      }), {
+        remotePeer: localPeer
+      }),
+      encrypter.secureOutbound(stubInterface<MultiaddrConnection>({
+        ...outbound
+      }), {
+        remotePeer: localPeer
+      })
+    ])
+
+    expect(result).to.have.nested.property('[0].streamMuxer.protocol', '/test/muxer')
+    expect(result).to.have.nested.property('[1].streamMuxer.protocol', '/test/muxer')
+  })
 })

packages/integration-tests/package.json

@@ -83,6 +83,7 @@
     "p-retry": "^6.2.1",
     "p-wait-for": "^5.0.2",
     "sinon": "^19.0.2",
+    "sinon-ts": "^2.0.0",
     "uint8arraylist": "^2.4.8",
     "uint8arrays": "^5.1.0",
     "wherearewe": "^2.0.1"

packages/integration-tests/test/compliance/connection-encryption/tls.spec.ts

@@ -4,7 +4,9 @@ import { generateKeyPair } from '@libp2p/crypto/keys'
 import suite from '@libp2p/interface-compliance-tests/connection-encryption'
 import { defaultLogger } from '@libp2p/logger'
 import { tls } from '@libp2p/tls'
+import { stubInterface } from 'sinon-ts'
 import { isBrowser, isWebWorker } from 'wherearewe'
+import type { StreamMuxerFactory, Upgrader } from '@libp2p/interface'
 
 describe('tls connection encrypter interface compliance', () => {
   if (isBrowser || isWebWorker) {
@@ -15,7 +17,12 @@ describe('tls connection encrypter interface compliance', () => {
     async setup (opts) {
       return tls()({
         privateKey: opts?.privateKey ?? await generateKeyPair('Ed25519'),
-        logger: defaultLogger()
+        logger: defaultLogger(),
+        upgrader: stubInterface<Upgrader>({
+          getStreamMuxers () {
+            return new Map([['/test/muxer', stubInterface<StreamMuxerFactory>()]])
+          }
+        })
       })
     },
     async teardown () {

packages/integration-tests/test/interop.ts

@@ -138,7 +138,7 @@ async function createJsPeer (options: SpawnOptions): Promise<Daemon> {
       webRTCDirect()
     ],
     streamMuxers: [],
-    connectionEncrypters: [noise()]
+    connectionEncrypters: []
   }
 
   if (options.noListen !== true) {
@@ -159,12 +159,12 @@ async function createJsPeer (options: SpawnOptions): Promise<Daemon> {
     identify: identify()
   }
 
-  if (options.encryption === 'noise') {
-    opts.connectionEncrypters?.push(noise())
-  } else if (options.encryption === 'tls') {
+  if (options.encryption === 'tls') {
     opts.connectionEncrypters?.push(tls())
   } else if (options.encryption === 'plaintext') {
     opts.connectionEncrypters?.push(plaintext())
+  } else {
+    opts.connectionEncrypters?.push(noise())
   }
 
   if (options.muxer === 'mplex') {