fix!: close streams gracefully

- Refactors `.close`, `closeRead` and `.closeWrite` methods on the `Stream` interface to be async
- The `Connection` interface now has `.close` and `.abort` methods
- `.close` on `Stream`s and `Connection`s wait for the internal message queues to empty before closing
- `.abort` on `Stream`s and `Connection`s close the underlying stream immediately and discards any unsent data
- `@chainsafe/libp2p-yamux` now uses the `AbstractStream` class from `@libp2p/interface` the same as `@libp2p/mplex` and
`@libp2p/webrtc`

Follow-up PRs will be necessary to `@chainsafe/libp2p-yamux`, `@chainsafe/libp2p-gossipsub` and `@chainsafe/libp2p-noise`
though they will not block the release as their code is temporarily added to this repo to let CI run.

Fixes #1793
Fixes #656

BREAKING CHANGE: the `.close`, `closeRead` and `closeWrite` methods on the `Stream` interface are now asynchronous

Author: achingbrain

doc/METRICS.md

@@ -65,7 +65,7 @@ const node = await createLibp2p({
 To define component metrics first get a reference to the metrics object:
 
 ```ts
-import type { Metrics } from '@libp2p/interface-metrics'
+import type { Metrics } from '@libp2p/interface/metrics'
 
 interface MyClassComponents {
   metrics: Metrics

packages/connection-encryption-noise/package.json

@@ -0,0 +1,99 @@
+{
+  "name": "@chainsafe/libp2p-noise",
+  "version": "12.0.1",
+  "author": "ChainSafe <[email protected]>",
+  "license": "Apache-2.0 OR MIT",
+  "homepage": "https://github.com/ChainSafe/js-libp2p-noise#readme",
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/ChainSafe/js-libp2p-noise.git"
+  },
+  "bugs": {
+    "url": "https://github.com/ChainSafe/js-libp2p-noise/issues"
+  },
+  "keywords": [
+    "crypto",
+    "libp2p",
+    "noise"
+  ],
+  "engines": {
+    "node": ">=16.0.0",
+    "npm": ">=7.0.0"
+  },
+  "type": "module",
+  "types": "./dist/src/index.d.ts",
+  "files": [
+    "src",
+    "dist",
+    "!dist/test",
+    "!**/*.tsbuildinfo"
+  ],
+  "exports": {
+    ".": {
+      "types": "./dist/src/index.d.ts",
+      "import": "./dist/src/index.js"
+    }
+  },
+  "eslintConfig": {
+    "extends": "ipfs",
+    "parserOptions": {
+      "sourceType": "module"
+    },
+    "rules": {
+      "@typescript-eslint/no-unused-vars": "error",
+      "@typescript-eslint/explicit-function-return-type": "warn",
+      "@typescript-eslint/strict-boolean-expressions": "off"
+    },
+    "ignorePatterns": [
+      "src/proto/payload.js",
+      "src/proto/payload.d.ts",
+      "test/fixtures/node-globals.js"
+    ]
+  },
+  "scripts": {
+    "bench": "node benchmarks/benchmark.js",
+    "clean": "aegir clean",
+    "dep-check": "aegir dep-check",
+    "build": "aegir build",
+    "lint": "aegir lint",
+    "lint:fix": "aegir lint --fix",
+    "test": "aegir test",
+    "test:node": "aegir test -t node",
+    "test:browser": "aegir test -t browser -t webworker",
+    "test:electron-main": "aegir test -t electron-main",
+    "docs": "aegir docs",
+    "proto:gen": "protons ./src/proto/payload.proto",
+    "prepublish": "npm run build"
+  },
+  "dependencies": {
+    "@libp2p/crypto": "^1.0.11",
+    "@libp2p/interface": "~0.0.1",
+    "@libp2p/logger": "^2.0.5",
+    "@libp2p/peer-id": "^2.0.0",
+    "@stablelib/chacha20poly1305": "^1.0.1",
+    "@noble/hashes": "^1.3.0",
+    "@stablelib/x25519": "^1.0.3",
+    "it-length-prefixed": "^9.0.1",
+    "it-length-prefixed-stream": "^1.0.0",
+    "it-byte-stream": "^1.0.0",
+    "it-pair": "^2.0.2",
+    "it-pipe": "^3.0.1",
+    "it-stream-types": "^2.0.1",
+    "protons-runtime": "^5.0.0",
+    "uint8arraylist": "^2.3.2",
+    "uint8arrays": "^4.0.2"
+  },
+  "devDependencies": {
+    "@libp2p/interface-compliance-tests": "^3.0.0",
+    "@libp2p/peer-id-factory": "^2.0.0",
+    "@types/sinon": "^10.0.14",
+    "aegir": "^39.0.5",
+    "iso-random-stream": "^2.0.2",
+    "protons": "^7.0.0",
+    "sinon": "^15.0.0"
+  },
+  "browser": {
+    "./dist/src/alloc-unsafe.js": "./dist/src/alloc-unsafe-browser.js",
+    "util": false
+  }
+}

packages/connection-encryption-noise/src/@types/basic.ts

@@ -0,0 +1,5 @@
+export type bytes = Uint8Array
+export type bytes32 = Uint8Array
+export type bytes16 = Uint8Array
+
+export type uint64 = number

packages/connection-encryption-noise/src/@types/handshake-interface.ts

@@ -0,0 +1,12 @@
+import type { bytes } from './basic.js'
+import type { NoiseSession } from './handshake.js'
+import type { NoiseExtensions } from '../proto/payload.js'
+import type { PeerId } from '@libp2p/interface/peer-id'
+
+export interface IHandshake {
+  session: NoiseSession
+  remotePeer: PeerId
+  remoteExtensions: NoiseExtensions
+  encrypt: (plaintext: bytes, session: NoiseSession) => bytes
+  decrypt: (ciphertext: bytes, session: NoiseSession, dst?: Uint8Array) => { plaintext: bytes, valid: boolean }
+}

packages/connection-encryption-noise/src/@types/handshake.ts

@@ -0,0 +1,48 @@
+import type { bytes, bytes32, uint64 } from './basic.js'
+import type { KeyPair } from './libp2p.js'
+import type { Nonce } from '../nonce.js'
+
+export type Hkdf = [bytes, bytes, bytes]
+
+export interface MessageBuffer {
+  ne: bytes32
+  ns: bytes
+  ciphertext: bytes
+}
+
+export interface CipherState {
+  k: bytes32
+  // For performance reasons, the nonce is represented as a Nonce object
+  // The nonce is treated as a uint64, even though the underlying `number` only has 52 safely-available bits.
+  n: Nonce
+}
+
+export interface SymmetricState {
+  cs: CipherState
+  ck: bytes32 // chaining key
+  h: bytes32 // handshake hash
+}
+
+export interface HandshakeState {
+  ss: SymmetricState
+  s: KeyPair
+  e?: KeyPair
+  rs: bytes32
+  re: bytes32
+  psk: bytes32
+}
+
+export interface NoiseSession {
+  hs: HandshakeState
+  h?: bytes32
+  cs1?: CipherState
+  cs2?: CipherState
+  mc: uint64
+  i: boolean
+}
+
+export interface INoisePayload {
+  identityKey: bytes
+  identitySig: bytes
+  data: bytes
+}

packages/connection-encryption-noise/src/@types/libp2p.ts

@@ -0,0 +1,10 @@
+import type { bytes32 } from './basic.js'
+import type { NoiseExtensions } from '../proto/payload.js'
+import type { ConnectionEncrypter } from '@libp2p/interface/connection-encrypter'
+
+export interface KeyPair {
+  publicKey: bytes32
+  privateKey: bytes32
+}
+
+export interface INoiseConnection extends ConnectionEncrypter<NoiseExtensions> {}

packages/connection-encryption-noise/src/constants.ts

@@ -0,0 +1,4 @@
+export const NOISE_MSG_MAX_LENGTH_BYTES = 65535
+export const NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG = NOISE_MSG_MAX_LENGTH_BYTES - 16
+
+export const DUMP_SESSION_KEYS = Boolean(globalThis.process?.env?.DUMP_SESSION_KEYS)

packages/connection-encryption-noise/src/crypto.ts

@@ -0,0 +1,16 @@
+import type { bytes32, bytes } from './@types/basic.js'
+import type { Hkdf } from './@types/handshake.js'
+import type { KeyPair } from './@types/libp2p.js'
+
+export interface ICryptoInterface {
+  hashSHA256: (data: Uint8Array) => Uint8Array
+
+  getHKDF: (ck: bytes32, ikm: Uint8Array) => Hkdf
+
+  generateX25519KeyPair: () => KeyPair
+  generateX25519KeyPairFromSeed: (seed: Uint8Array) => KeyPair
+  generateX25519SharedKey: (privateKey: Uint8Array, publicKey: Uint8Array) => Uint8Array
+
+  chaCha20Poly1305Encrypt: (plaintext: Uint8Array, nonce: Uint8Array, ad: Uint8Array, k: bytes32) => bytes
+  chaCha20Poly1305Decrypt: (ciphertext: Uint8Array, nonce: Uint8Array, ad: Uint8Array, k: bytes32, dst?: Uint8Array) => bytes | null
+}

packages/connection-encryption-noise/src/crypto/js.ts

@@ -0,0 +1,58 @@
+import { hkdf } from '@noble/hashes/hkdf'
+import { sha256 } from '@noble/hashes/sha256'
+import { ChaCha20Poly1305 } from '@stablelib/chacha20poly1305'
+import * as x25519 from '@stablelib/x25519'
+import type { bytes, bytes32 } from '../@types/basic.js'
+import type { Hkdf } from '../@types/handshake.js'
+import type { KeyPair } from '../@types/libp2p.js'
+import type { ICryptoInterface } from '../crypto.js'
+
+export const pureJsCrypto: ICryptoInterface = {
+  hashSHA256 (data: Uint8Array): Uint8Array {
+    return sha256(data)
+  },
+
+  getHKDF (ck: bytes32, ikm: Uint8Array): Hkdf {
+    const okm = hkdf(sha256, ikm, ck, undefined, 96)
+
+    const k1 = okm.subarray(0, 32)
+    const k2 = okm.subarray(32, 64)
+    const k3 = okm.subarray(64, 96)
+
+    return [k1, k2, k3]
+  },
+
+  generateX25519KeyPair (): KeyPair {
+    const keypair = x25519.generateKeyPair()
+
+    return {
+      publicKey: keypair.publicKey,
+      privateKey: keypair.secretKey
+    }
+  },
+
+  generateX25519KeyPairFromSeed (seed: Uint8Array): KeyPair {
+    const keypair = x25519.generateKeyPairFromSeed(seed)
+
+    return {
+      publicKey: keypair.publicKey,
+      privateKey: keypair.secretKey
+    }
+  },
+
+  generateX25519SharedKey (privateKey: Uint8Array, publicKey: Uint8Array): Uint8Array {
+    return x25519.sharedKey(privateKey, publicKey)
+  },
+
+  chaCha20Poly1305Encrypt (plaintext: Uint8Array, nonce: Uint8Array, ad: Uint8Array, k: bytes32): bytes {
+    const ctx = new ChaCha20Poly1305(k)
+
+    return ctx.seal(nonce, plaintext, ad)
+  },
+
+  chaCha20Poly1305Decrypt (ciphertext: Uint8Array, nonce: Uint8Array, ad: Uint8Array, k: bytes32, dst?: Uint8Array): bytes | null {
+    const ctx = new ChaCha20Poly1305(k)
+
+    return ctx.open(nonce, ciphertext, ad, dst)
+  }
+}

packages/connection-encryption-noise/src/crypto/streaming.ts

@@ -0,0 +1,58 @@
+import { TAG_LENGTH } from '@stablelib/chacha20poly1305'
+import { NOISE_MSG_MAX_LENGTH_BYTES, NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG } from '../constants.js'
+import { uint16BEEncode } from '../encoder.js'
+import type { IHandshake } from '../@types/handshake-interface.js'
+import type { MetricsRegistry } from '../metrics.js'
+import type { Transform } from 'it-stream-types'
+import type { Uint8ArrayList } from 'uint8arraylist'
+
+// Returns generator that encrypts payload from the user
+export function encryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<AsyncIterable<Uint8Array>> {
+  return async function * (source) {
+    for await (const chunk of source) {
+      for (let i = 0; i < chunk.length; i += NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG) {
+        let end = i + NOISE_MSG_MAX_LENGTH_BYTES_WITHOUT_TAG
+        if (end > chunk.length) {
+          end = chunk.length
+        }
+
+        const data = handshake.encrypt(chunk.subarray(i, end), handshake.session)
+        metrics?.encryptedPackets.increment()
+
+        yield uint16BEEncode(data.byteLength)
+        yield data
+      }
+    }
+  }
+}
+
+// Decrypt received payload to the user
+export function decryptStream (handshake: IHandshake, metrics?: MetricsRegistry): Transform<AsyncIterable<Uint8ArrayList>, AsyncIterable<Uint8Array>> {
+  return async function * (source) {
+    for await (const chunk of source) {
+      for (let i = 0; i < chunk.length; i += NOISE_MSG_MAX_LENGTH_BYTES) {
+        let end = i + NOISE_MSG_MAX_LENGTH_BYTES
+        if (end > chunk.length) {
+          end = chunk.length
+        }
+
+        if (end - TAG_LENGTH < i) {
+          throw new Error('Invalid chunk')
+        }
+        const encrypted = chunk.subarray(i, end)
+        // memory allocation is not cheap so reuse the encrypted Uint8Array
+        // see https://github.com/ChainSafe/js-libp2p-noise/pull/242#issue-1422126164
+        // this is ok because chacha20 reads bytes one by one and don't reread after that
+        // it's also tested in https://github.com/ChainSafe/as-chacha20poly1305/pull/1/files#diff-25252846b58979dcaf4e41d47b3eadd7e4f335e7fb98da6c049b1f9cd011f381R48
+        const dst = chunk.subarray(i, end - TAG_LENGTH)
+        const { plaintext: decrypted, valid } = handshake.decrypt(encrypted, handshake.session, dst)
+        if (!valid) {
+          metrics?.decryptErrors.increment()
+          throw new Error('Failed to validate decrypted chunk')
+        }
+        metrics?.decryptedPackets.increment()
+        yield decrypted
+      }
+    }
+  }
+}