feat: add webtransport listener

.github/workflows/main.yml

@@ -46,6 +46,10 @@ jobs:
       fail-fast: true
     steps:
       - uses: actions/checkout@v4
+      # needed for webtransport tests, remove after https://github.com/libp2p/js-libp2p/pull/2422
+      - uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: '1.22'
       - uses: actions/setup-node@v4
         with:
           node-version: ${{ matrix.node }}
@@ -147,6 +151,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      # needed for webtransport tests, remove after https://github.com/libp2p/js-libp2p/pull/2422
+      - uses: actions/setup-go@0c52d547c9bc32b1aa3301fd7a9cb496313a4491 # v5.0.0
+        with:
+          go-version: '1.22'
       - uses: actions/setup-node@v4
         with:
           node-version: lts/*

packages/integration-tests/package.json

@@ -48,7 +48,7 @@
     "@libp2p/interface": "^1.3.1",
     "@libp2p/interface-compliance-tests": "^5.4.4",
     "@libp2p/interface-internal": "^1.2.1",
-    "@libp2p/interop": "^11.0.0",
+    "@libp2p/interop": "^12.0.2",
     "@libp2p/kad-dht": "^12.0.16",
     "@libp2p/logger": "^4.0.12",
     "@libp2p/mdns": "^10.0.23",

packages/integration-tests/test/interop.ts

@@ -15,6 +15,7 @@ import { mplex } from '@libp2p/mplex'
 import { peerIdFromKeys } from '@libp2p/peer-id'
 import { tcp } from '@libp2p/tcp'
 import { tls } from '@libp2p/tls'
+import { webTransport } from '@libp2p/webtransport'
 import { multiaddr } from '@multiformats/multiaddr'
 import { execa } from 'execa'
 import { path as p2pd } from 'go-libp2p'
@@ -46,7 +47,11 @@ async function createGoPeer (options: SpawnOptions): Promise<Daemon> {
   if (options.noListen === true) {
     opts.push('-noListenAddrs')
   } else {
-    opts.push('-hostAddrs=/ip4/127.0.0.1/tcp/0')
+    if (options.transport === 'webtransport') {
+      opts.push('-hostAddrs=/ip4/127.0.0.1/udp/0/quic-v1/webtransport')
+    } else {
+      opts.push('-hostAddrs=/ip4/127.0.0.1/tcp/0')
+    }
   }
 
   if (options.encryption != null) {
@@ -122,16 +127,22 @@ async function createJsPeer (options: SpawnOptions): Promise<Daemon> {
   const opts: Libp2pOptions<ServiceMap> = {
     peerId,
     addresses: {
-      listen: options.noListen === true ? [] : ['/ip4/127.0.0.1/tcp/0']
+      listen: []
     },
-    transports: [tcp(), circuitRelayTransport()],
+    transports: [tcp(), circuitRelayTransport(), webTransport()],
     streamMuxers: [],
     connectionEncryption: [noise()],
     connectionManager: {
       minConnections: 0
     }
   }
 
+  if (options.transport === 'webtransport') {
+    opts.addresses?.listen?.push('/ip4/127.0.0.1/udp/0/quic-v1/webtransport')
+  } else if (options.transport === 'tcp') {
+    opts.addresses?.listen?.push('/ip4/127.0.0.1/tcp/0')
+  }
+
   const services: ServiceFactoryMap = {
     identify: identify()
   }

packages/transport-webtransport/.aegir.js

@@ -1,5 +1,3 @@
-import { createClient } from '@libp2p/daemon-client'
-import { multiaddr } from '@multiformats/multiaddr'
 import { execa } from 'execa'
 import { path as p2pd } from 'go-libp2p'
 import pDefer from 'p-defer'
@@ -27,6 +25,12 @@ export default {
 }
 
 async function createGoLibp2p () {
+  // have to import these dynamically as they have a transitive dependency on
+  // @libp2p/interface which is part of this monorepo so may not have been built
+  // yet
+  const { multiaddr } = await import('@multiformats/multiaddr')
+  const { createClient } = await import('@libp2p/daemon-client')
+
   const controlPort = Math.floor(Math.random() * (50000 - 10000 + 1)) + 10000
   const apiAddr = multiaddr(`/ip4/127.0.0.1/tcp/${controlPort}`)
   const deferred = pDefer()

packages/transport-webtransport/package.json

@@ -42,30 +42,36 @@
   "scripts": {
     "clean": "aegir clean",
     "lint": "aegir lint",
-    "dep-check": "aegir dep-check",
+    "dep-check": "aegir dep-check -i @fails-components/webtransport-transport-http3-quiche",
     "doc-check": "aegir doc-check",
     "build": "aegir build",
-    "test": "aegir test -t browser -t webworker",
+    "test": "aegir test",
+    "test:node": "aegir test -t node --cov",
     "test:chrome": "aegir test -t browser --cov",
     "test:chrome-webworker": "aegir test -t webworker"
   },
   "dependencies": {
     "@chainsafe/libp2p-noise": "^15.0.0",
+    "@fails-components/webtransport": "^1.1.0",
+    "@fails-components/webtransport-transport-http3-quiche": "^1.1.0",
     "@libp2p/interface": "^1.3.1",
     "@libp2p/peer-id": "^4.1.1",
     "@libp2p/utils": "^5.4.1",
     "@multiformats/multiaddr": "^12.2.3",
     "@multiformats/multiaddr-matcher": "^1.2.1",
+    "@peculiar/x509": "^1.9.7",
+    "browser-readablestream-to-it": "^2.0.5",
     "it-stream-types": "^2.0.1",
     "multiformats": "^13.1.0",
     "race-signal": "^1.0.2",
     "uint8arraylist": "^2.4.8",
     "uint8arrays": "^5.1.0"
   },
   "devDependencies": {
-    "@libp2p/daemon-client": "^8.0.5",
+    "@libp2p/interface-compliance-tests": "^5.4.4",
     "@libp2p/logger": "^4.0.12",
     "@libp2p/peer-id-factory": "^4.1.1",
+    "@libp2p/daemon-client": "^8.0.5",
     "@libp2p/ping": "^1.0.18",
     "@noble/hashes": "^1.4.0",
     "aegir": "^42.2.11",
@@ -75,7 +81,8 @@
     "it-to-buffer": "^4.0.7",
     "libp2p": "^1.5.2",
     "p-defer": "^4.0.1",
-    "p-wait-for": "^5.0.2"
+    "p-wait-for": "^5.0.2",
+    "sinon": "^18.0.0"
   },
   "browser": {
     "./dist/src/listener.js": "./dist/src/listener.browser.js",

packages/transport-webtransport/src/create-server.ts

@@ -0,0 +1,150 @@
+import { Http3Server } from '@fails-components/webtransport'
+import { TypedEventEmitter } from '@libp2p/interface'
+import { raceSignal } from 'race-signal'
+import type { HttpServerInit, WebTransportSession } from '@fails-components/webtransport'
+import type { ComponentLogger, Logger, TypedEventTarget } from '@libp2p/interface'
+
+interface WebTransportServerEvents extends Record<string, any> {
+  listening: CustomEvent
+  session: CustomEvent<WebTransportSession>
+  close: CustomEvent
+  error: CustomEvent<Error>
+}
+
+export interface WebTransportServer extends TypedEventTarget<WebTransportServerEvents> {
+  listening: boolean
+  sessionTimeout: number
+
+  close(callback?: () => void): void
+  listen(): void
+  address(): { port: number, host: string, family: 'IPv4' | 'IPv6' } | null
+}
+
+export interface WebTransportServerComponents {
+  logger: ComponentLogger
+}
+
+class DefaultWebTransportServer extends TypedEventEmitter<WebTransportServerEvents> implements WebTransportServer {
+  private readonly server: Http3Server
+  public listening: boolean
+  /**
+   * How long in ms to wait for an incoming session to be ready
+   */
+  public sessionTimeout: number
+  private readonly log: Logger
+
+  constructor (components: WebTransportServerComponents, init: HttpServerInit) {
+    super()
+
+    this.server = new Http3Server(init)
+    this.listening = false
+    this.log = components.logger.forComponent('libp2p:webtransport:server')
+
+    this.sessionTimeout = 1000
+  }
+
+  close (callback?: () => void): void {
+    if (callback != null) {
+      this.addEventListener('close', callback)
+    }
+
+    this.server.stopServer()
+    this.server.closed
+      .then(() => {
+        this.listening = false
+        this.safeDispatchEvent('close')
+      })
+      .catch((err) => {
+        this.safeDispatchEvent('error', { detail: err })
+      })
+  }
+
+  listen (): void {
+    this.server.startServer()
+    this.server.ready
+      .then(() => {
+        this.server.setRequestCallback(async (args: any): Promise<any> => {
+          const url = args.header[':path']
+          const [path] = url.split('?')
+
+          if (this.server.sessionController[path] == null) {
+            return {
+              ...args,
+              path,
+              status: 404
+            }
+          }
+
+          return {
+            ...args,
+            path,
+            userData: {
+              search: url.substring(path.length)
+            },
+            header: {
+              ...args.header,
+              ':path': path
+            },
+            status: 200
+          }
+        })
+
+        this.listening = true
+        this.safeDispatchEvent('listening')
+
+        this.log('ready, processing incoming sessions')
+        this._processIncomingSessions()
+          .catch(err => {
+            this.safeDispatchEvent('error', { detail: err })
+          })
+      })
+      .catch((err) => {
+        this.safeDispatchEvent('error', { detail: err })
+      })
+  }
+
+  address (): { port: number, host: string, family: 'IPv4' | 'IPv6' } | null {
+    return this.server.address()
+  }
+
+  async _processIncomingSessions (): Promise<void> {
+    const sessionStream = this.server.sessionStream('/.well-known/libp2p-webtransport')
+    const sessionReader = sessionStream.getReader()
+
+    while (true) {
+      const { done, value: session } = await sessionReader.read()
+
+      if (done) {
+        this.log('session reader finished')
+        break
+      }
+
+      this.log('new incoming session')
+      void Promise.resolve()
+        .then(async () => {
+          try {
+            await raceSignal(session.ready, AbortSignal.timeout(this.sessionTimeout))
+            this.log('session ready')
+
+            this.safeDispatchEvent('session', { detail: session })
+          } catch (err) {
+            this.log.error('error waiting for session to become ready', err)
+          }
+        })
+    }
+  }
+}
+
+export interface SessionHandler {
+  (event: CustomEvent<WebTransportSession>): void
+}
+
+export function createServer (components: WebTransportServerComponents, init: HttpServerInit, sessionHandler?: SessionHandler): WebTransportServer {
+  const server = new DefaultWebTransportServer(components, init)
+
+  if (sessionHandler != null) {
+    server.addEventListener('session', sessionHandler)
+  }
+
+  return server
+}

packages/transport-webtransport/src/index.ts

@@ -191,6 +191,10 @@
         throw new CodeError('Failed to authenticate webtransport', 'ERR_AUTHENTICATION_FAILED')
       }
 
+      if (options?.signal?.aborted === true) {
+        throw new AbortError()
+      }
+
       this.metrics?.dialerEvents.increment({ open: true })
 
       maConn = {
@@ -303,8 +307,8 @@
   /**
    * Filter check for all Multiaddrs that this transport can listen on
    */
-  listenFilter (): Multiaddr[] {
-    return []
+  listenFilter (multiaddrs: Multiaddr[]): Multiaddr[] {
+    return multiaddrs.filter(ma => WebTransportMatcher.exactMatch(ma))
   }
 
   /**