fix(ssh): harden remote-forward routing and accept-loop teardown

Key remote forwards by bind host and port (not port alone) so two
forwards sharing a port stay distinct; guard the accept loops against a
failing CHANNEL_OPEN_CONFIRMATION send (close the upstream, keep the loop
alive); reject pending RemoteForward.accept() waiters when the connection
closes or its mux fails.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

packages/ssh/src/connection.test.ts

@@ -122,4 +122,38 @@ suite("ssh connections with forwarding", () => {
       await close()
     }
   })
+
+  test("two forwards sharing a port on different hosts stay distinct", async () => {
+    const [listenerA, dialerA] = loopbackListener()
+    const [listenerB, dialerB] = loopbackListener()
+    const { client, close } = await connected({
+      tcpipForward: async (host) => (host === "127.0.0.1" ? listenerA : listenerB),
+    })
+    try {
+      const fwdA = await client.requestRemoteForward("127.0.0.1", 9000)
+      const fwdB = await client.requestRemoteForward("10.0.0.1", 9000)
+
+      const acceptB = fwdB.accept()
+      const inboundB = await dialerB.dial("10.0.0.1", 9000)
+      const forwardedB = await acceptB
+      void echo(forwardedB)
+      const payload = new Uint8Array([7, 7, 7])
+      await inboundB.write(payload)
+      assert.deepEqual(await readN(inboundB, 3), payload)
+
+      // Closing fwdB must not disturb fwdA: a connection on A still routes.
+      await inboundB.close()
+      await fwdB.close()
+      const acceptA = fwdA.accept()
+      const inboundA = await dialerA.dial("127.0.0.1", 9000)
+      const forwardedA = await acceptA
+      void echo(forwardedA)
+      await inboundA.write(payload)
+      assert.deepEqual(await readN(inboundA, 3), payload)
+      await inboundA.close()
+      await fwdA.close()
+    } finally {
+      await close()
+    }
+  })
 })

packages/ssh/src/connection.ts

@@ -23,7 +23,7 @@ function parseAddr(addr: string | undefined): TcpipEndpoint {
 
 export class SSHClientConnection {
   readonly #mux: ConnectionMux
-  readonly #forwards = new Map<number, RemoteForward>()
+  readonly #forwards = new Map<string, RemoteForward>()
   #forwardLoop = false
 
   private constructor(mux: ConnectionMux) {
@@ -82,12 +82,22 @@ export class SSHClientConnection {
       new Writer().string(bindHost).u32(bindPort).finish(),
     )
     const port = bindPort === 0 ? new Reader(reply).u32() : bindPort
-    const forward = new RemoteForward(this.#mux, bindHost, port, () => this.#forwards.delete(port))
-    this.#forwards.set(port, forward)
+    const key = `${bindHost}:${port}`
+    const forward = new RemoteForward(this.#mux, bindHost, port, () => this.#forwards.delete(key))
+    this.#forwards.set(key, forward)
     this.#startForwardLoop()
     return forward
   }
 
+  #matchForward(bound: TcpipEndpoint): RemoteForward | undefined {
+    const exact = this.#forwards.get(`${bound.host}:${bound.port}`)
+    if (exact) return exact
+    // OpenSSH may report the bound address differently than requested
+    // (e.g. "" vs "localhost"); fall back to a unique port match.
+    for (const forward of this.#forwards.values()) if (forward.port === bound.port) return forward
+    return undefined
+  }
+
   #startForwardLoop(): void {
     if (this.#forwardLoop) return
     this.#forwardLoop = true
@@ -100,21 +110,34 @@ export class SSHClientConnection {
       try {
         open = await this.#mux.acceptOpen()
       } catch {
+        this.#shutdownForwards()
         return
       }
-      const forward = open.bound ? this.#forwards.get(open.bound.port) : undefined
-      if (open.type !== "forwarded-tcpip" || !forward) {
+      const forward =
+        open.type === "forwarded-tcpip" && open.bound ? this.#matchForward(open.bound) : undefined
+      if (!forward) {
         await open.reject(SSH_OPEN.ADMINISTRATIVELY_PROHIBITED)
         continue
       }
-      const channel = await open.accept()
+      let channel: Channel
+      try {
+        channel = await open.accept()
+      } catch {
+        continue
+      }
       const origin = open.originator ?? { host: "127.0.0.1", port: 0 }
       forward._deliver(channelTransport(channel, { remoteAddr: `${origin.host}:${origin.port}` }))
     }
   }
 
-  close(): Promise<void> {
-    return this.#mux.close()
+  #shutdownForwards(): void {
+    for (const forward of this.#forwards.values()) forward._shutdown()
+    this.#forwards.clear()
+  }
+
+  async close(): Promise<void> {
+    this.#shutdownForwards()
+    await this.#mux.close()
   }
 }
 
@@ -148,6 +171,13 @@ export class RemoteForward implements RawListener {
     else this.#queue.push(transport)
   }
 
+  _shutdown(): void {
+    if (this.#closed) return
+    this.#closed = true
+    for (const w of this.#waiters) w.reject(new Error("connection closed"))
+    this.#waiters = []
+  }
+
   accept(): Promise<RawTransport> {
     const queued = this.#queue.shift()
     if (queued) return Promise.resolve(queued)
@@ -227,7 +257,12 @@ export class SSHServerConnection<T> {
         return
       }
       if (open.type === "session") {
-        const channel = await open.accept()
+        let channel: Channel
+        try {
+          channel = await open.accept()
+        } catch {
+          continue
+        }
         const w = this.#sessionWaiters.shift()
         if (w) w.resolve(channel)
         else this.#sessionQueue.push(channel)
@@ -252,7 +287,13 @@ export class SSHServerConnection<T> {
       await open.reject(SSH_OPEN.CONNECT_FAILED)
       return
     }
-    const channel = await open.accept()
+    let channel: Channel
+    try {
+      channel = await open.accept()
+    } catch {
+      await closeQuietly(upstream)
+      return
+    }
     void pipe(channelTransport(channel), upstream)
   }