fix: support validating asymmetric addresses (#2515)

Some transports can listen on addresses that validate differently
to ones that they dial.

For example WebTransport requires cert hashes to dial but can generate
them and add them to the listened-on address when listening.

Splits the `.filter` method into `dialFilter` and `listenFilter`.

Author: achingbrain

packages/interface-compliance-tests/src/transport/dial-test.ts

@@ -17,7 +17,8 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
   describe('dial', () => {
     let upgrader: Upgrader
     let registrar: Registrar
-    let addrs: Multiaddr[]
+    let listenAddrs: Multiaddr[]
+    let dialAddrs: Multiaddr[]
     let transport: Transport
     let connector: Connector
     let listener: Listener
@@ -29,7 +30,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         events: new TypedEventEmitter()
       });
 
-      ({ addrs, transport, connector } = await common.setup())
+      ({ listenAddrs, dialAddrs, transport, connector } = await common.setup())
     })
 
     after(async () => {
@@ -40,7 +41,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       listener = transport.createListener({
         upgrader
       })
-      await listener.listen(addrs[0])
+      await listener.listen(listenAddrs[0])
     })
 
     afterEach(async () => {
@@ -61,7 +62,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       })
 
       const upgradeSpy = sinon.spy(upgrader, 'upgradeOutbound')
-      const conn = await transport.dial(addrs[0], {
+      const conn = await transport.dial(dialAddrs[0], {
         upgrader
       })
 
@@ -77,7 +78,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
 
     it('can close connections', async () => {
       const upgradeSpy = sinon.spy(upgrader, 'upgradeOutbound')
-      const conn = await transport.dial(addrs[0], {
+      const conn = await transport.dial(dialAddrs[0], {
         upgrader
       })
 
@@ -90,7 +91,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
     it('to non existent listener', async () => {
       const upgradeSpy = sinon.spy(upgrader, 'upgradeOutbound')
 
-      await expect(transport.dial(addrs[1], {
+      await expect(transport.dial(dialAddrs[1], {
         upgrader
       })).to.eventually.be.rejected()
       expect(upgradeSpy.callCount).to.equal(0)
@@ -100,7 +101,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       const upgradeSpy = sinon.spy(upgrader, 'upgradeOutbound')
       const controller = new AbortController()
       controller.abort()
-      const conn = transport.dial(addrs[0], { signal: controller.signal, upgrader })
+      const conn = transport.dial(dialAddrs[0], { signal: controller.signal, upgrader })
 
       await expect(conn).to.eventually.be.rejected().with.property('code', AbortError.code)
       expect(upgradeSpy.callCount).to.equal(0)
@@ -113,7 +114,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       connector.delay(100)
 
       const controller = new AbortController()
-      const conn = transport.dial(addrs[0], { signal: controller.signal, upgrader })
+      const conn = transport.dial(dialAddrs[0], { signal: controller.signal, upgrader })
       setTimeout(() => { controller.abort() }, 50)
 
       await expect(conn).to.eventually.be.rejected().with.property('code', AbortError.code)

packages/interface-compliance-tests/src/transport/filter-test.ts

@@ -6,20 +6,26 @@ import type { Multiaddr } from '@multiformats/multiaddr'
 
 export default (common: TestSetup<TransportTestFixtures>): void => {
   describe('filter', () => {
-    let addrs: Multiaddr[]
+    let listenAddrs: Multiaddr[]
+    let dialAddrs: Multiaddr[]
     let transport: Transport
 
     before(async () => {
-      ({ addrs, transport } = await common.setup())
+      ({ listenAddrs, dialAddrs, transport } = await common.setup())
     })
 
     after(async () => {
       await common.teardown()
     })
 
-    it('filters addresses', () => {
-      const filteredAddrs = transport.filter(addrs)
-      expect(filteredAddrs).to.eql(addrs)
+    it('filters listen addresses', () => {
+      const filteredAddrs = transport.listenFilter(listenAddrs)
+      expect(filteredAddrs).to.eql(listenAddrs)
+    })
+
+    it('filters dial addresses', () => {
+      const filteredAddrs = transport.dialFilter(dialAddrs)
+      expect(filteredAddrs).to.eql(dialAddrs)
     })
   })
 }

packages/interface-compliance-tests/src/transport/index.ts

@@ -11,7 +11,8 @@ export interface Connector {
 }
 
 export interface TransportTestFixtures {
-  addrs: Multiaddr[]
+  listenAddrs: Multiaddr[]
+  dialAddrs: Multiaddr[]
   transport: Transport
   connector: Connector
 }

packages/interface-compliance-tests/src/transport/listen-test.ts

@@ -18,7 +18,8 @@ import type { Multiaddr } from '@multiformats/multiaddr'
 export default (common: TestSetup<TransportTestFixtures>): void => {
   describe('listen', () => {
     let upgrader: Upgrader
-    let addrs: Multiaddr[]
+    let listenAddrs: Multiaddr[]
+    let dialAddrs: Multiaddr[]
     let transport: Transport
     let registrar: Registrar
 
@@ -29,7 +30,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         events: new TypedEventEmitter()
       });
 
-      ({ transport, addrs } = await common.setup())
+      ({ transport, listenAddrs, dialAddrs } = await common.setup())
     })
 
     after(async () => {
@@ -44,7 +45,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       const listener = transport.createListener({
         upgrader
       })
-      await listener.listen(addrs[0])
+      await listener.listen(listenAddrs[0])
       await listener.close()
     })
 
@@ -65,14 +66,14 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       })
 
       // Listen
-      await listener.listen(addrs[0])
+      await listener.listen(listenAddrs[0])
 
       // Create two connections to the listener
       const [conn1] = await Promise.all([
-        transport.dial(addrs[0], {
+        transport.dial(dialAddrs[0], {
           upgrader
         }),
-        transport.dial(addrs[0], {
+        transport.dial(dialAddrs[0], {
           upgrader
         })
       ])
@@ -112,10 +113,10 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
       })
 
       // Listen
-      await listener.listen(addrs[0])
+      await listener.listen(listenAddrs[0])
 
       // Create a connection to the listener
-      const conn = await transport.dial(addrs[0], {
+      const conn = await transport.dial(dialAddrs[0], {
         upgrader
       })
 
@@ -138,8 +139,8 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         })
 
         void (async () => {
-          await listener.listen(addrs[0])
-          await transport.dial(addrs[0], {
+          await listener.listen(listenAddrs[0])
+          await transport.dial(dialAddrs[0], {
             upgrader
           })
         })()
@@ -158,7 +159,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         listener.addEventListener('listening', () => {
           listener.close().then(done, done)
         })
-        void listener.listen(addrs[0])
+        void listener.listen(listenAddrs[0])
       })
 
       it('error', (done) => {
@@ -181,7 +182,7 @@ export default (common: TestSetup<TransportTestFixtures>): void => {
         listener.addEventListener('close', () => { done() })
 
         void (async () => {
-          await listener.listen(addrs[0])
+          await listener.listen(listenAddrs[0])
           await listener.close()
         })()
       })

packages/interface-internal/src/transport-manager/index.ts

@@ -31,9 +31,15 @@ export interface TransportManager {
   getListeners(): Listener[]
 
   /**
-   * Get the transport for a given multiaddr, if one has been configured
+   * Get the transport to dial a given multiaddr, if one has been configured
    */
-  transportForMultiaddr(ma: Multiaddr): Transport | undefined
+  dialTransportForMultiaddr(ma: Multiaddr): Transport | undefined
+
+  /**
+   * Get the transport to listen on a given multiaddr, if one has been
+   * configured
+   */
+  listenTransportForMultiaddr(ma: Multiaddr): Transport | undefined
 
   /**
    * Listen on the passed multiaddrs

packages/interface/src/transport/index.ts

@@ -68,9 +68,16 @@ export interface Transport {
   createListener(options: CreateListenerOptions): Listener
 
   /**
-   * Takes a list of `Multiaddr`s and returns only valid addresses for the transport
+   * Takes a list of `Multiaddr`s and returns only addresses that are valid for
+   * the transport to listen on
    */
-  filter: MultiaddrFilter
+  listenFilter: MultiaddrFilter
+
+  /**
+   * Takes a list of `Multiaddr`s and returns only addresses that are vali for
+   * the transport to dial
+   */
+  dialFilter: MultiaddrFilter
 }
 
 export function isTransport (other: any): other is Transport {

packages/libp2p/src/connection-manager/dial-queue.ts

@@ -398,7 +398,7 @@ export class DialQueue {
 
     const filteredAddrs = resolvedAddresses.filter(addr => {
       // filter out any multiaddrs that we do not have transports for
-      if (this.components.transportManager.transportForMultiaddr(addr.multiaddr) == null) {
+      if (this.components.transportManager.dialTransportForMultiaddr(addr.multiaddr) == null) {
         return false
       }
 

packages/libp2p/src/transport-manager.ts

@@ -106,7 +106,7 @@ export class DefaultTransportManager implements TransportManager, Startable {
    * Dials the given Multiaddr over it's supported transport
    */
   async dial (ma: Multiaddr, options?: AbortOptions): Promise<Connection> {
-    const transport = this.transportForMultiaddr(ma)
+    const transport = this.dialTransportForMultiaddr(ma)
 
     if (transport == null) {
       throw new CodeError(`No transport available for address ${String(ma)}`, codes.ERR_TRANSPORT_UNAVAILABLE)
@@ -156,9 +156,22 @@ export class DefaultTransportManager implements TransportManager, Startable {
   /**
    * Finds a transport that matches the given Multiaddr
    */
-  transportForMultiaddr (ma: Multiaddr): Transport | undefined {
+  dialTransportForMultiaddr (ma: Multiaddr): Transport | undefined {
     for (const transport of this.transports.values()) {
-      const addrs = transport.filter([ma])
+      const addrs = transport.dialFilter([ma])
+
+      if (addrs.length > 0) {
+        return transport
+      }
+    }
+  }
+
+  /**
+   * Finds a transport that matches the given Multiaddr
+   */
+  listenTransportForMultiaddr (ma: Multiaddr): Transport | undefined {
+    for (const transport of this.transports.values()) {
+      const addrs = transport.listenFilter([ma])
 
       if (addrs.length > 0) {
         return transport
@@ -182,7 +195,7 @@ export class DefaultTransportManager implements TransportManager, Startable {
     const couldNotListen = []
 
     for (const [key, transport] of this.transports.entries()) {
-      const supportedAddrs = transport.filter(addrs)
+      const supportedAddrs = transport.listenFilter(addrs)
       const tasks = []
 
       // For each supported multiaddr, create a listener

packages/libp2p/test/connection-manager/dial-queue.spec.ts

@@ -57,7 +57,7 @@ describe('dial queue', () => {
       '/ip4/127.0.0.1/tcp/1233': async () => deferredConn.promise
     }
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.callsFake(async ma => {
       const maStr = ma.toString()
       const action = actions[maStr]
@@ -97,7 +97,7 @@ describe('dial queue', () => {
       ]
     })
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.withArgs(matchMultiaddr(ma.encapsulate(`/p2p/${peerId}`))).resolves(connection)
 
     dialer = new DialQueue(components)
@@ -124,7 +124,7 @@ describe('dial queue', () => {
       ]
     })
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.withArgs(matchMultiaddr(ma.encapsulate(`/p2p/${peerId}`))).resolves(connection)
 
     dialer = new DialQueue(components)
@@ -141,7 +141,7 @@ describe('dial queue', () => {
       '/ip4/127.0.0.1/tcp/1233': async () => deferredConn.promise
     }
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.callsFake(async ma => {
       const maStr = ma.toString()
       const action = actions[maStr]
@@ -178,7 +178,7 @@ describe('dial queue', () => {
       maxParallelDials: 2
     })
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.callsFake(async ma => {
       const maStr = ma.toString()
       const action = actions[maStr]
@@ -218,7 +218,7 @@ describe('dial queue', () => {
       maxParallelDials: 2
     })
 
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
     components.transportManager.dial.callsFake(async ma => {
       const maStr = ma.toString()
       const action = actions[maStr]
@@ -267,7 +267,7 @@ describe('dial queue', () => {
     dialer = new DialQueue(components, {
       maxParallelDials: 50
     })
-    components.transportManager.transportForMultiaddr.returns(stubInterface<Transport>())
+    components.transportManager.dialTransportForMultiaddr.returns(stubInterface<Transport>())
 
     const connection = mockConnection(mockMultiaddrConnection(mockDuplex(), remotePeer))
 
@@ -294,7 +294,7 @@ describe('dial queue', () => {
     const ma = multiaddr(`/ip4/123.123.123.123/tcp/123/ws/p2p/${relayPeer}/p2p-circuit/webrtc`)
     const maWithPeer = `${ma}/p2p/${remotePeer}`
 
-    components.transportManager.transportForMultiaddr.callsFake(ma => {
+    components.transportManager.dialTransportForMultiaddr.callsFake(ma => {
       if (WebRTC.exactMatch(ma)) {
         return stubInterface<Transport>()
       }

packages/libp2p/test/connection-manager/direct.spec.ts

@@ -507,7 +507,7 @@ describe('libp2p.dialer (direct, WebSockets)', () => {
 
   it('should limit the maximum dial queue size', async () => {
     const transport = stubInterface<Transport>({
-      filter: (ma) => ma,
+      dialFilter: (ma) => ma,
       dial: async () => {
         await delay(1000)
         return stubInterface<Connection>()