feat(ssh): add client and server connections with TCP forwarding

SSHClientConnection/SSHServerConnection wrap the transport, auth, and
channel mux; expose direct-tcpip local forwarding (dialer()) and
tcpip-forward remote forwarding (RemoteForward as a RawListener).

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

packages/ssh/src/_internals.ts

@@ -8,3 +8,4 @@ export * from "./transport.js"
 export * from "./auth.js"
 export * from "./channel.js"
 export * from "./forward.js"
+export * from "./connection.js"

packages/ssh/src/channel.ts

@@ -19,19 +19,15 @@ export type ConnectionMuxOptions = {
   onGlobalRequest?: GlobalRequestHandler
 }
 
-type IncomingOpenBase = {
+export type IncomingOpen = {
+  type: string
+  dest?: TcpipEndpoint
+  bound?: TcpipEndpoint
+  originator?: TcpipEndpoint
   accept(): Promise<Channel>
   reject(code?: number, description?: string): Promise<void>
 }
 
-export type IncomingOpen = IncomingOpenBase &
-  (
-    | { type: "session" }
-    | { type: "direct-tcpip"; dest: TcpipEndpoint; originator: TcpipEndpoint }
-    | { type: "forwarded-tcpip"; bound: TcpipEndpoint; originator: TcpipEndpoint }
-    | { type: string }
-  )
-
 export class Channel {
   readonly #t: SSHTransport
   readonly #localId: number
@@ -508,8 +504,8 @@ export class ConnectionMux {
     window: number,
     maxPacket: number,
   ): IncomingOpen {
-    const base: IncomingOpenBase = {
-      accept: async () => {
+    const base = {
+      accept: async (): Promise<Channel> => {
         const localId = this.#nextId++
         const ch = new Channel(this.#t, localId)
         ch._confirmOutbound(senderId, window, maxPacket)

packages/ssh/src/connection.test.ts

@@ -0,0 +1,125 @@
+import { suite, test } from "node:test"
+import assert from "node:assert/strict"
+import {
+  loopbackTransportPair,
+  loopbackListener,
+  type LoopbackTransportHalf,
+} from "@webnet/transport/loopback"
+import type { RawTransport } from "@webnet/transport"
+import { generateHostKey } from "./keys.js"
+import { SSHClientConnection, SSHServerConnection } from "./connection.js"
+import type { SSHServerConnectionOptions } from "./connection.js"
+
+async function connected(serverOpts: Partial<SSHServerConnectionOptions<object>> = {}): Promise<{
+  client: SSHClientConnection
+  server: SSHServerConnection<object>
+  close: () => Promise<void>
+}> {
+  const [a, b] = loopbackTransportPair()
+  const hostKey = await generateHostKey()
+  const [client, server] = await Promise.all([
+    SSHClientConnection.connect(a, { user: "u", password: "p" }),
+    SSHServerConnection.accept(b, {
+      hostKey,
+      authenticate: async () => ({}),
+      ...serverOpts,
+    }),
+  ])
+  return {
+    client,
+    server,
+    close: async () => {
+      await client.close().catch(() => {})
+      await server.close().catch(() => {})
+    },
+  }
+}
+
+async function echo(t: RawTransport): Promise<void> {
+  for (;;) {
+    let chunk: Uint8Array
+    try {
+      chunk = await t.read()
+    } catch {
+      return
+    }
+    await t.write(chunk).catch(() => {})
+  }
+}
+
+async function readN(t: RawTransport, n: number): Promise<Uint8Array> {
+  const out = new Uint8Array(n)
+  let off = 0
+  while (off < n) {
+    const chunk = await t.read()
+    out.set(chunk, off)
+    off += chunk.length
+  }
+  return out
+}
+
+suite("ssh connections with forwarding", () => {
+  test("local forward: dialer() reaches the directTcpip hook", async () => {
+    let seen: { host: string; port: number } | null = null
+    const { client, close } = await connected({
+      directTcpip: async (dest) => {
+        seen = dest
+        const [near, far] = loopbackTransportPair()
+        void echo(far)
+        return near
+      },
+    })
+    try {
+      const conn = client.dialer()
+      const t = await conn.dial("service.internal", 8080)
+      assert.deepEqual(seen, { host: "service.internal", port: 8080 })
+      const payload = new Uint8Array([9, 8, 7, 6])
+      await t.write(payload)
+      assert.deepEqual(await readN(t, 4), payload)
+      await t.close()
+    } finally {
+      await close()
+    }
+  })
+
+  test("directTcpip returning null rejects the dial", async () => {
+    const { client, close } = await connected({ directTcpip: async () => null })
+    try {
+      await assert.rejects(async () => {
+        const t = await client.dialer().dial("nope", 1)
+        await t.read()
+      }, /channel open failed/)
+    } finally {
+      await close()
+    }
+  })
+
+  test("remote forward: tcpipForward listener receives connections", async () => {
+    const [listener, dialer] = loopbackListener()
+    const { client, close } = await connected({
+      tcpipForward: async (host, port) => {
+        assert.equal(host, "127.0.0.1")
+        assert.equal(port, 9000)
+        return listener
+      },
+    })
+    try {
+      const forward = await client.requestRemoteForward("127.0.0.1", 9000)
+      assert.equal(forward.port, 9000)
+
+      // A connection arriving on the server-side listener surfaces on the client.
+      const acceptP = forward.accept()
+      const inbound = (await dialer.dial("127.0.0.1", forward.port)) as LoopbackTransportHalf
+      const forwarded = await acceptP
+      void echo(forwarded)
+
+      const payload = new Uint8Array([1, 2, 3, 4, 5])
+      await inbound.write(payload)
+      assert.deepEqual(await readN(inbound, 5), payload)
+      await inbound.close()
+      await forward.close()
+    } finally {
+      await close()
+    }
+  })
+})

packages/ssh/src/connection.ts

@@ -0,0 +1,331 @@
+import type { RawDialer, RawListener, RawTransport } from "@webnet/transport"
+import { Reader, Writer } from "./cursor.js"
+import { SSHTransport, type HostKeyVerifier } from "./transport.js"
+import { importSigningKey, parsePrivateKey } from "./keys.js"
+import { authenticateClient, authenticateServer, type Credential } from "./auth.js"
+import { SSH_OPEN } from "./constants.js"
+import { ConnectionMux, type Channel, type IncomingOpen, type TcpipEndpoint } from "./channel.js"
+import { channelTransport, pipe } from "./forward.js"
+
+export type SSHClientConnectionOptions = {
+  user: string
+  password?: string
+  privateKey?: string
+  verifyHostKey?: HostKeyVerifier
+}
+
+function parseAddr(addr: string | undefined): TcpipEndpoint {
+  if (!addr) return { host: "127.0.0.1", port: 0 }
+  const i = addr.lastIndexOf(":")
+  if (i < 0) return { host: addr, port: 0 }
+  return { host: addr.slice(0, i), port: Number(addr.slice(i + 1)) || 0 }
+}
+
+export class SSHClientConnection {
+  readonly #mux: ConnectionMux
+  readonly #forwards = new Map<number, RemoteForward>()
+  #forwardLoop = false
+
+  private constructor(mux: ConnectionMux) {
+    this.#mux = mux
+  }
+
+  static async connect(
+    raw: RawTransport,
+    opts: SSHClientConnectionOptions,
+  ): Promise<SSHClientConnection> {
+    const transport = await SSHTransport.create(raw, {
+      role: "client",
+      verifyHostKey: opts.verifyHostKey,
+    })
+    let key: { publicKey: Uint8Array<ArrayBuffer>; signingKey: CryptoKey } | undefined
+    if (opts.privateKey) {
+      const parsed = parsePrivateKey(opts.privateKey)
+      key = { publicKey: parsed.publicKey, signingKey: await importSigningKey(parsed.privateSeed) }
+    }
+    await authenticateClient(transport, {
+      user: opts.user,
+      password: opts.password,
+      privateKey: key,
+    })
+    const mux = new ConnectionMux(transport, { acceptTypes: ["forwarded-tcpip"] })
+    mux.start()
+    return new SSHClientConnection(mux)
+  }
+
+  async openSubsystem(name: string): Promise<Channel> {
+    const channel = await this.#mux.openSession()
+    await channel.requestSubsystem(name)
+    return channel
+  }
+
+  openSession(): Promise<Channel> {
+    return this.#mux.openSession()
+  }
+
+  openDirectTcpip(dest: TcpipEndpoint, originator?: TcpipEndpoint): Promise<Channel> {
+    return this.#mux.openDirectTcpip(dest, originator ?? { host: "127.0.0.1", port: 0 })
+  }
+
+  dialer(): RawDialer {
+    return {
+      dial: async (host: string, port: number): Promise<RawTransport> => {
+        const channel = await this.openDirectTcpip({ host, port })
+        return channelTransport(channel, { remoteAddr: `${host}:${port}` })
+      },
+    }
+  }
+
+  async requestRemoteForward(bindHost: string, bindPort: number): Promise<RemoteForward> {
+    const reply = await this.#mux.globalRequest(
+      "tcpip-forward",
+      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)
+    this.#startForwardLoop()
+    return forward
+  }
+
+  #startForwardLoop(): void {
+    if (this.#forwardLoop) return
+    this.#forwardLoop = true
+    void this.#forwardLoop_()
+  }
+
+  async #forwardLoop_(): Promise<void> {
+    for (;;) {
+      let open: IncomingOpen
+      try {
+        open = await this.#mux.acceptOpen()
+      } catch {
+        return
+      }
+      const forward = open.bound ? this.#forwards.get(open.bound.port) : undefined
+      if (open.type !== "forwarded-tcpip" || !forward) {
+        await open.reject(SSH_OPEN.ADMINISTRATIVELY_PROHIBITED)
+        continue
+      }
+      const channel = await open.accept()
+      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()
+  }
+}
+
+export class RemoteForward implements RawListener {
+  readonly #mux: ConnectionMux
+  readonly bindHost: string
+  readonly port: number
+  readonly #onClose: () => void
+  #closed = false
+  #queue: RawTransport[] = []
+  #waiters: { resolve: (t: RawTransport) => void; reject: (e: unknown) => void }[] = []
+
+  constructor(mux: ConnectionMux, bindHost: string, port: number, onClose: () => void) {
+    this.#mux = mux
+    this.bindHost = bindHost
+    this.port = port
+    this.#onClose = onClose
+  }
+
+  get addr(): string {
+    return `${this.bindHost}:${this.port}`
+  }
+
+  get closed(): boolean {
+    return this.#closed
+  }
+
+  _deliver(transport: RawTransport): void {
+    const w = this.#waiters.shift()
+    if (w) w.resolve(transport)
+    else this.#queue.push(transport)
+  }
+
+  accept(): Promise<RawTransport> {
+    const queued = this.#queue.shift()
+    if (queued) return Promise.resolve(queued)
+    if (this.#closed) return Promise.reject(new Error("forward closed"))
+    return new Promise((resolve, reject) => this.#waiters.push({ resolve, reject }))
+  }
+
+  async close(): Promise<void> {
+    if (this.#closed) return
+    this.#closed = true
+    this.#onClose()
+    for (const w of this.#waiters) w.reject(new Error("forward closed"))
+    this.#waiters = []
+    await this.#mux
+      .globalRequest(
+        "cancel-tcpip-forward",
+        new Writer().string(this.bindHost).u32(this.port).finish(),
+      )
+      .catch(() => {})
+  }
+}
+
+export type SSHServerConnectionOptions<T> = {
+  hostKey?: string
+  authenticate: (user: string, credential: Credential) => Promise<T | null>
+  directTcpip?: (dest: TcpipEndpoint, originator: TcpipEndpoint) => Promise<RawTransport | null>
+  tcpipForward?: (bindHost: string, bindPort: number) => Promise<RawListener | null>
+  maxAuthAttempts?: number
+}
+
+export class SSHServerConnection<T> {
+  readonly #mux: ConnectionMux
+  readonly #opts: SSHServerConnectionOptions<T>
+  readonly #listeners = new Map<string, RawListener>()
+  #sessionQueue: Channel[] = []
+  #sessionWaiters: { resolve: (c: Channel) => void; reject: (e: unknown) => void }[] = []
+  readonly user: string
+  readonly context: T
+
+  private constructor(
+    transport: SSHTransport,
+    opts: SSHServerConnectionOptions<T>,
+    user: string,
+    context: T,
+  ) {
+    this.#opts = opts
+    this.user = user
+    this.context = context
+    this.#mux = new ConnectionMux(transport, {
+      acceptTypes: opts.directTcpip ? ["session", "direct-tcpip"] : ["session"],
+      onGlobalRequest: (name, payload) => this.#onGlobalRequest(name, payload),
+    })
+  }
+
+  static async accept<T>(
+    raw: RawTransport,
+    opts: SSHServerConnectionOptions<T>,
+  ): Promise<SSHServerConnection<T>> {
+    const transport = await SSHTransport.create(raw, { role: "server", hostKey: opts.hostKey })
+    const { user, context } = await authenticateServer(transport, {
+      authenticate: opts.authenticate,
+      maxAttempts: opts.maxAuthAttempts,
+    })
+    const conn = new SSHServerConnection(transport, opts, user, context)
+    conn.#mux.start()
+    void conn.#loop()
+    return conn
+  }
+
+  async #loop(): Promise<void> {
+    for (;;) {
+      let open: IncomingOpen
+      try {
+        open = await this.#mux.acceptOpen()
+      } catch (err) {
+        this.#failSessions(err)
+        return
+      }
+      if (open.type === "session") {
+        const channel = await open.accept()
+        const w = this.#sessionWaiters.shift()
+        if (w) w.resolve(channel)
+        else this.#sessionQueue.push(channel)
+      } else if (open.type === "direct-tcpip") {
+        void this.#handleDirectTcpip(open)
+      } else {
+        await open.reject(SSH_OPEN.ADMINISTRATIVELY_PROHIBITED)
+      }
+    }
+  }
+
+  async #handleDirectTcpip(open: IncomingOpen): Promise<void> {
+    const dest = open.dest ?? { host: "", port: 0 }
+    const originator = open.originator ?? { host: "", port: 0 }
+    let upstream: RawTransport | null
+    try {
+      upstream = (await this.#opts.directTcpip?.(dest, originator)) ?? null
+    } catch {
+      upstream = null
+    }
+    if (!upstream) {
+      await open.reject(SSH_OPEN.CONNECT_FAILED)
+      return
+    }
+    const channel = await open.accept()
+    void pipe(channelTransport(channel), upstream)
+  }
+
+  async #onGlobalRequest(name: string, payload: Reader): Promise<Uint8Array<ArrayBuffer> | null> {
+    if (name === "tcpip-forward" && this.#opts.tcpipForward) {
+      const bindHost = payload.utf8()
+      const bindPort = payload.u32()
+      const listener = await this.#opts.tcpipForward(bindHost, bindPort)
+      if (!listener) return null
+      const port = bindPort === 0 ? parseAddr(listener.addr).port : bindPort
+      if (bindPort === 0 && !port) {
+        await listener.close()
+        return null
+      }
+      this.#listeners.set(`${bindHost}:${port}`, listener)
+      void this.#forwardAccept(listener, bindHost, port)
+      return bindPort === 0 ? new Writer().u32(port).finish() : new Uint8Array(0)
+    }
+    if (name === "cancel-tcpip-forward") {
+      const bindHost = payload.utf8()
+      const bindPort = payload.u32()
+      const key = `${bindHost}:${bindPort}`
+      const listener = this.#listeners.get(key)
+      if (!listener) return null
+      this.#listeners.delete(key)
+      await listener.close()
+      return new Uint8Array(0)
+    }
+    return null
+  }
+
+  async #forwardAccept(listener: RawListener, bindHost: string, port: number): Promise<void> {
+    for (;;) {
+      let inbound: RawTransport
+      try {
+        inbound = await listener.accept()
+      } catch {
+        return
+      }
+      try {
+        const channel = await this.#mux.openForwardedTcpip(
+          { host: bindHost, port },
+          parseAddr(inbound.remoteAddr),
+        )
+        void pipe(channelTransport(channel), inbound)
+      } catch {
+        await closeQuietly(inbound)
+      }
+    }
+  }
+
+  acceptSession(): Promise<Channel> {
+    const queued = this.#sessionQueue.shift()
+    if (queued) return Promise.resolve(queued)
+    return new Promise((resolve, reject) => this.#sessionWaiters.push({ resolve, reject }))
+  }
+
+  #failSessions(err: unknown): void {
+    for (const w of this.#sessionWaiters) w.reject(err)
+    this.#sessionWaiters = []
+  }
+
+  async close(): Promise<void> {
+    for (const listener of this.#listeners.values()) await closeQuietly(listener)
+    this.#listeners.clear()
+    await this.#mux.close()
+  }
+}
+
+async function closeQuietly(x: { close(): void | Promise<void> }): Promise<void> {
+  try {
+    await x.close()
+  } catch {
+    /* already closed */
+  }
+}

packages/ssh/src/forward.ts

@@ -35,7 +35,10 @@ export function channelTransport(
       return channel.closeSend()
     },
     close(): Promise<void> {
-      return channel.close()
+      // Close our side without blocking on the peer's close handshake, as a
+      // RawTransport.close() is expected to return promptly.
+      void channel.close()
+      return Promise.resolve()
     },
   }
 }

packages/ssh/src/index.ts

@@ -2,5 +2,8 @@ export { generateHostKey } from "./keys.js"
 export { Channel } from "./channel.js"
 export type { TcpipEndpoint, IncomingOpen } from "./channel.js"
 export { channelTransport } from "./forward.js"
+export { SSHClientConnection, SSHServerConnection, RemoteForward } from "./connection.js"
+export type { SSHClientConnectionOptions, SSHServerConnectionOptions } from "./connection.js"
 export { SSHAuthError } from "./auth.js"
 export type { Credential } from "./auth.js"
+export type { HostKeyInfo, HostKeyVerifier } from "./transport.js"