From 891fc3dc16588a6dcb1a5622f877d9d3d625c085 Mon Sep 17 00:00:00 2001 From: sovtech Date: Thu, 22 Oct 2026 17:35:34 -0600 Subject: [PATCH 1/5] build(seeder): pin sodium-universal and @hyperswarm/secret-stream, and type the NFX-12 surface - `sodium-universal` 5.0.1 becomes a dependency of @sovit/seeder, for the NFX-12 `bind` signature (Ed25519 by the peer key). `@hyperswarm/secret-stream` 6.9.1 becomes a devDependency, for in-memory secret-stream pairs in the transport's tests. Both were already in the lockfile at those versions (core and hyperdht pull them in): the lockfile diff is the two dependency lines of the workspace entry. No native module moves: sodium-native's shortest chain is still app-desktop's own dependency. - Ambient types, written against the installed sources: hyperdht 6.34.0's server and connect (`createServer`, `listen`, `connect`, `fullyBootstrapped`), secret-stream 6.9.1 (`setKeepAlive`, `setTimeout`, `opened`, pause/resume), protomux 3.11.0 (`from`, `createChannel`, `pair`, channels and messages), and the two sodium members the seeder calls. --- package-lock.json | 2 + packages/seeder/package.json | 2 + packages/seeder/src/types/holepunch.d.ts | 139 ++++++++++++++++++++++- packages/seeder/src/types/sodium.d.ts | 15 +++ 4 files changed, 154 insertions(+), 4 deletions(-) create mode 100644 packages/seeder/src/types/sodium.d.ts diff --git a/package-lock.json b/package-lock.json index 77258555..8f8cd7f5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10628,9 +10628,11 @@ "hyperswarm": "4.17.0", "nostr-tools": "2.25.2", "protomux": "3.11.0", + "sodium-universal": "5.0.1", "ws": "8.21.3" }, "devDependencies": { + "@hyperswarm/secret-stream": "6.9.1", "@types/ws": "8.18.1", "hyperdht": "6.34.0" } diff --git a/packages/seeder/package.json b/packages/seeder/package.json index 09b8c364..ccc577f3 100644 --- a/packages/seeder/package.json +++ b/packages/seeder/package.json @@ -31,9 +31,11 @@ "hyperswarm": "4.17.0", "nostr-tools": "2.25.2", "protomux": "3.11.0", + "sodium-universal": "5.0.1", "ws": "8.21.3" }, "devDependencies": { + "@hyperswarm/secret-stream": "6.9.1", "@types/ws": "8.18.1", "hyperdht": "6.34.0" } diff --git a/packages/seeder/src/types/holepunch.d.ts b/packages/seeder/src/types/holepunch.d.ts index 87b0165d..7c1e50e7 100644 --- a/packages/seeder/src/types/holepunch.d.ts +++ b/packages/seeder/src/types/holepunch.d.ts @@ -1,8 +1,9 @@ /** * Minimal ambient declarations for the Holepunch stack, covering EXACTLY the surface the * seeder uses. Written against the pinned versions (hypercore@11.35.3, hyperblobs@2.12.1, - * corestore@7.12.2, hyperswarm@4.17.0, hyperdht@6.34.0) by reading their source — see - * docs/archive/lanes/L2.md. Nothing here is a full typing of those libraries. + * corestore@7.12.2, hyperswarm@4.17.0, hyperdht@6.34.0, protomux@3.11.0, + * @hyperswarm/secret-stream@6.9.1) by reading their source — see docs/archive/lanes/L2.md. + * Nothing here is a full typing of those libraries. */ declare module 'hypercore' { @@ -232,14 +233,144 @@ declare module 'hyperswarm' { } declare module 'hyperdht' { + import type NoiseSecretStream from '@hyperswarm/secret-stream'; + + export interface KeyPair { + publicKey: Uint8Array; + secretKey: Uint8Array; + } + + export interface DHTOptions { + readonly bootstrap?: readonly { host: string; port: number }[]; + /** UDP bind host; `127.0.0.1` keeps the node off every other interface. */ + readonly host?: string; + readonly ephemeral?: boolean; + } + + export interface ServerOptions { + /** + * Called with the remote's Noise key during the handshake (`lib/server.js` `_addHandshake`); + * return `true` to REJECT. Only the first argument is used here. + */ + readonly firewall?: (remotePublicKey: Uint8Array) => boolean | Promise; + } + + /** `lib/server.js` `Server`: listens on one key pair. */ + export interface Server { + readonly publicKey: Uint8Array | null; + listen(keyPair: KeyPair): Promise; + close(): Promise; + } + + export interface ConnectOptions { + readonly keyPair?: KeyPair; + } + class DHT { - constructor(opts?: { readonly bootstrap?: readonly { host: string; port: number }[] }); - static keyPair(seed?: Uint8Array): { publicKey: Uint8Array; secretKey: Uint8Array }; + constructor(opts?: DHTOptions); + static keyPair(seed?: Uint8Array): KeyPair; + fullyBootstrapped(): Promise; + /** Every accepted connection is a secret-stream whose `remotePublicKey` is the dialer's. */ + createServer(opts: ServerOptions, onconnection: (socket: NoiseSecretStream) => void): Server; + /** The returned stream opens once the Noise IK handshake with `remotePublicKey` completes. */ + connect(remotePublicKey: Uint8Array, opts?: ConnectOptions): NoiseSecretStream; destroy(): Promise; } export default DHT; } +declare module '@hyperswarm/secret-stream' { + /** The paired raw side of a stream made without a transport (an in-memory Bridge). */ + export interface RawStream { + pipe(dest: T): T; + } + + /** + * A streamx Duplex of framed, encrypted messages. `write` returns `false` once 16 KiB are + * buffered (streamx `highWaterMark`), and `'drain'` fires when the buffer has emptied. + */ + class NoiseSecretStream { + constructor( + isInitiator: boolean, + rawStream?: null, + opts?: { readonly keyPair?: { publicKey: Uint8Array; secretKey: Uint8Array } }, + ); + readonly isInitiator: boolean; + readonly publicKey: Uint8Array | null; + readonly remotePublicKey: Uint8Array | null; + readonly handshakeHash: Uint8Array | null; + /** Resolves `true` once the handshake is done, `false` if the stream closed first. */ + readonly opened: Promise; + readonly destroyed: boolean; + readonly rawStream: RawStream; + /** Protomux keeps itself here (`Protomux.from`). */ + userData: unknown; + write(data: Uint8Array): boolean; + destroy(err?: Error): void; + pause(): this; + resume(): this; + /** Send an empty keepalive frame after `ms` without a write. */ + setKeepAlive(ms: number): void; + /** Destroy the stream after `ms` without receiving anything (refreshed by every frame). */ + setTimeout(ms: number): void; + on(event: 'open' | 'close' | 'drain' | 'end', cb: () => void): this; + on(event: 'error', cb: (err: Error) => void): this; + once(event: 'open' | 'close' | 'drain', cb: () => void): this; + once(event: 'error', cb: (err: Error) => void): this; + off(event: 'open' | 'close' | 'drain', cb: () => void): this; + off(event: 'error', cb: (err: Error) => void): this; + } + export default NoiseSecretStream; +} + +declare module 'protomux' { + /** `addMessage` result: `send` returns the stream's `write()` (`false` = wait for drain). */ + export interface ProtomuxMessage { + send(message: Uint8Array): boolean; + } + + /** `Channel` (protomux 3.11.0 `index.js`). Messages are numbered in `addMessage` order. */ + export interface ProtomuxChannel { + readonly protocol: string; + readonly id: Uint8Array | null; + readonly opened: boolean; + readonly closed: boolean; + /** No encoding → compact-encoding `raw`: the handler gets a view into the frame. */ + addMessage(opts: { readonly onmessage?: (message: Uint8Array) => unknown }): ProtomuxMessage; + open(): void; + close(): void; + /** `true` once the remote opened its side, `false` if it rejected or the channel closed. */ + fullyOpened(): Promise; + } + + export interface ChannelOptions { + readonly protocol: string; + readonly id?: Uint8Array | null; + readonly onopen?: () => unknown; + readonly onclose?: (isRemote: boolean) => unknown; + } + + class Protomux { + /** The stream's existing muxer (`stream.userData`), or a new one. */ + static from(stream: unknown): Protomux; + /** + * `null` when the stream is destroyed, or when a channel for (protocol, id) is already + * open on this mux (`unique`, the default). + */ + createChannel(opts: ChannelOptions): ProtomuxChannel | null; + /** + * `notify(id)` runs when the remote opens a channel of `protocol` that is not open here + * (`id: null` = any id without a pairing of its own). It is awaited; if it creates no + * channel for that id, the remote's open is rejected. + */ + pair( + opts: { readonly protocol: string; readonly id?: Uint8Array | null }, + notify: (id: Uint8Array | null) => unknown, + ): void; + } + export default Protomux; +} + declare module 'hyperdht/testnet.js' { export interface Testnet { readonly bootstrap: readonly { host: string; port: number }[]; diff --git a/packages/seeder/src/types/sodium.d.ts b/packages/seeder/src/types/sodium.d.ts new file mode 100644 index 00000000..028f7d43 --- /dev/null +++ b/packages/seeder/src/types/sodium.d.ts @@ -0,0 +1,15 @@ +/** + * Ambient types for `sodium-universal@5.0.1` (→ `sodium-native@5.1.0` on Node and Bare), only + * the members the seeder calls: the NFX-12 `bind` signature (`src/nfx/bind.ts`), RFC 8032 + * Ed25519 over a HyperDHT key pair's libsodium-format secret key. + */ +declare module 'sodium-universal' { + interface SodiumUniversal { + readonly crypto_sign_BYTES: number; + readonly crypto_sign_PUBLICKEYBYTES: number; + readonly crypto_sign_SECRETKEYBYTES: number; + crypto_sign_detached(sig: Uint8Array, m: Uint8Array, sk: Uint8Array): void; + } + const sodium: SodiumUniversal; + export default sodium; +} From ffe88b26e03b9fe39e7894be723ce346fdc54bf0 Mon Sep 17 00:00:00 2001 From: sovtech Date: Thu, 22 Oct 2026 17:35:47 -0600 Subject: [PATCH 2/5] =?UTF-8?q?feat(seeder/nfx):=20the=20NFX-12=20HyperDHT?= =?UTF-8?q?=20transport=20=E2=80=94=20dial,=20listen,=20files/1=20both=20s?= =?UTF-8?q?ides,=20pay/1=20lane?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transport of nfx `spec/nfx-12-hyperdht.md` (ADR 0020), free mode only, on the wire layouts of `@sovit/core`'s `nfx` codec. Nothing in the app uses it yet; the worker wires it up later. - `listenNfx` / `dialNfx` on a HyperDHT node the caller owns (bootstrap from local config only): the seeder listens on its per-network peer key, the watcher dials it with its own; the dialer closes a connection whose remote key is not the one it dialed. Local policy caps connections in total and per remote key, in the DHT firewall and again on arrival. - `NfxConnection`: keepalive at 25 s and close after 60 s of silence (secret-stream's timers); at most 16 videos per connection; a remote channel only for a canonical video address the source serves; 64 rejected opens close the peer (protomux 3.11 bounds neither pending nor rejected opens). - `nfx/files/1` seeder side: only files the source lists for THIS channel's video (never an oracle for the rest of the store), `missing` otherwise; at most 8 open requests per channel; malformed or misdirected messages, a reused `req` and trapped unknown types close the channel. - Flow control: one pump per connection writes one frame at a time and waits for `'drain'` after the stream reports full, so at most one 64 KiB frame sits on a buffer under 16 KiB, inside the 256 KiB cap; pay/1 is written directly, behind no more than that. Each frame is read inside the pump's turn, so a connection buffers one chunk however many requests it has. - `nfx/files/1` watcher side: each payload copied out of the receive buffer, capped at the file's size, sha256-checked before `ok`; a mismatch, too-long or too-short file drops the connection (NFX-05 §4). Sink writes are serialised and reads held above 1 MiB waiting. A request with no frame for 60 s is abandoned, unless the silence is the sink's. - `hyperdhtBind` / `hyperdhtEndpoint`: the beacon endpoint and its Ed25519 `bind`. - `PayLane`: `nfx/pay/1` carrying NFX-07 messages under the codec's rules, no payment logic. Protomux drops a message type a channel did not add without telling it, so after the pinned types each channel adds four trap types that close it; indices past those are still dropped unseen (no state changes), short of the close NFX-12 asks for. --- CHANGELOG.md | 7 + packages/seeder/README.md | 4 + packages/seeder/src/nfx/bind.ts | 67 +++++ packages/seeder/src/nfx/connection.ts | 238 +++++++++++++++++ packages/seeder/src/nfx/endpoint.ts | 176 +++++++++++++ packages/seeder/src/nfx/files-client.ts | 333 ++++++++++++++++++++++++ packages/seeder/src/nfx/files-server.ts | 187 +++++++++++++ packages/seeder/src/nfx/index.ts | 43 +++ packages/seeder/src/nfx/pay-lane.ts | 96 +++++++ packages/seeder/src/nfx/pump.ts | 132 ++++++++++ packages/seeder/src/nfx/traps.ts | 15 ++ packages/seeder/src/portable.ts | 43 +++ 12 files changed, 1341 insertions(+) create mode 100644 packages/seeder/src/nfx/bind.ts create mode 100644 packages/seeder/src/nfx/connection.ts create mode 100644 packages/seeder/src/nfx/endpoint.ts create mode 100644 packages/seeder/src/nfx/files-client.ts create mode 100644 packages/seeder/src/nfx/files-server.ts create mode 100644 packages/seeder/src/nfx/index.ts create mode 100644 packages/seeder/src/nfx/pay-lane.ts create mode 100644 packages/seeder/src/nfx/pump.ts create mode 100644 packages/seeder/src/nfx/traps.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a4263479..336ffb7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ All notable changes to the Nutflix desktop app are recorded here. The format fol its `bind`, hash lists and playlists, canonical JSON, and the `nfx/files/1` and `nfx/pay/1` message layouts. Every case of the suite's vectors runs in the tests, against copies pinned by sha256 in `docs/vendor/nfx-test-vectors/`. Nothing in the app uses it yet. +- `@sovit/seeder` speaks the NFX-12 HyperDHT transport (`nfx/`): a seeder listens on its + per-network peer key and serves whole files over `nfx/files/1`, a watcher dials that key and + fetches them, each file capped at its size and checked against its sha256 before it counts. + Flow control keeps at most one 64 KiB frame queued on top of the stream's buffer, and the + per-connection limits, keepalives and the 60-second idle rule are enforced. It also signs the + beacon's `hyperdht` `bind` and carries `nfx/pay/1` messages without acting on them. Free mode + only, and nothing in the app uses it yet. ### Changed diff --git a/packages/seeder/README.md b/packages/seeder/README.md index 7f99994a..f500f3ad 100644 --- a/packages/seeder/README.md +++ b/packages/seeder/README.md @@ -18,3 +18,7 @@ and the hardened systemd unit are documented in The standalone daemon serves the core it already holds. The command line has no option to load a video into it yet, so for now it is of use mainly as a building block: the desktop app embeds the same code in its worker. + +It also carries the NFX-12 HyperDHT transport (`src/nfx/`, ADR 0020): `listenNfx` serves whole +files by sha256 on a per-network peer key, `dialNfx` fetches them and checks each one. It is free +mode only for now, runs beside the Hypercore path above and shares no keys or stores with it. diff --git a/packages/seeder/src/nfx/bind.ts b/packages/seeder/src/nfx/bind.ts new file mode 100644 index 00000000..1d26436e --- /dev/null +++ b/packages/seeder/src/nfx/bind.ts @@ -0,0 +1,67 @@ +/** + * The NFX-12 §2 beacon endpoint `{t:"hyperdht", peer, bind}`: `bind` is the peer key's + * Ed25519 signature over `"nfx/hyperdht/1:" + seeder + ":" + namespace`, so a beacon can only + * name a peer key whose holder agreed to be advertised by that nostr key on that network. It + * depends on neither the video nor the time: a seeder signs it once per network. The message + * and the verifier are the codec's (`@sovit/core` `nfx`); only the signing lives here, beside + * the key pair the transport listens on. + */ +import { nfx } from '@sovit/core'; +import sodium from 'sodium-universal'; + +import { bytesEqual, toHex } from '../util/hex.js'; + +/** A HyperDHT (Noise) key pair: libsodium Ed25519, the secret key's last 32 bytes are the public key. */ +export interface PeerKeyPair { + readonly publicKey: Uint8Array; + readonly secretKey: Uint8Array; +} + +export interface HyperdhtEndpoint { + readonly t: 'hyperdht'; + readonly peer: string; + readonly bind: string; +} + +/** + * `bind` (128 lowercase hex) for `keyPair` advertised by `seeder` (64 lowercase hex nostr key) + * on `namespace` (NFX-01). Throws on a malformed argument: these are the caller's own values. + */ +export function hyperdhtBind(keyPair: PeerKeyPair, seeder: string, namespace: string): string { + checkKeyPair(keyPair); + if (!/^[0-9a-f]{64}$/.test(seeder)) { + throw new Error('invalid-argument: seeder is not 64 lowercase hex'); + } + const ns = nfx.Namespace.parse(namespace); + if (!ns.ok || ns.value.toString() !== namespace) { + throw new Error('invalid-argument: namespace is not an NFX-01 namespace'); + } + const sig = new Uint8Array(sodium.crypto_sign_BYTES); + const message = new TextEncoder().encode(nfx.hyperdhtBindMessage(seeder, namespace)); + sodium.crypto_sign_detached(sig, message, keyPair.secretKey); + return toHex(sig); +} + +/** The endpoint object for beacon content (NFX-03 §4), in the codec's field order. */ +export function hyperdhtEndpoint( + keyPair: PeerKeyPair, + seeder: string, + namespace: string, +): HyperdhtEndpoint { + return { + t: 'hyperdht', + peer: toHex(keyPair.publicKey), + bind: hyperdhtBind(keyPair, seeder, namespace), + }; +} + +function checkKeyPair(keyPair: PeerKeyPair): void { + const { publicKey, secretKey } = keyPair; + if ( + publicKey.byteLength !== sodium.crypto_sign_PUBLICKEYBYTES || + secretKey.byteLength !== sodium.crypto_sign_SECRETKEYBYTES || + !bytesEqual(secretKey.subarray(32), publicKey) + ) { + throw new Error('invalid-argument: not an Ed25519 key pair'); + } +} diff --git a/packages/seeder/src/nfx/connection.ts b/packages/seeder/src/nfx/connection.ts new file mode 100644 index 00000000..916b0533 --- /dev/null +++ b/packages/seeder/src/nfx/connection.ts @@ -0,0 +1,238 @@ +/** + * One NFX-12 connection: a HyperDHT secret-stream, its Protomux, and the per-connection rules + * of §3–§4. Dialing and listening are `endpoint.ts`; this owns what both sides share. + * + * - Liveness (§3): a keepalive after 25 s without a write, and the stream is destroyed after + * 60 s without receiving anything (secret-stream's own timers; empty frames are keepalives). + * - Flow control (§4): one `FilePump` for every `nfx/files/1` channel on the connection. + * - Limits (§4): at most 16 videos (so 32 channels) per connection. On the seeder side a + * remote channel is accepted only for a well-formed video address the source serves; one + * video too many closes the connection. On the watcher side `files`/`pay` refuse to open + * past the limit. + * - Channel ids are `utf8(:)` (§3, `VideoAddr.channelId()`), and an id + * that is not the canonical form of a valid address is rejected, not normalized. + */ +import type NoiseSecretStream from '@hyperswarm/secret-stream'; +import { nfx } from '@sovit/core'; +import Protomux from 'protomux'; + +import type { SeederCrypto } from '../adapters/crypto.js'; +import { FilesClient } from './files-client.js'; +import { serveFiles, type FilesServerOptions } from './files-server.js'; +import { PayLane, type PayLaneHandlers } from './pay-lane.js'; +import { FilePump } from './pump.js'; + +export interface NfxLimits { + /** At most 25 s (NFX-12 §3). */ + readonly keepAliveMs: number; + /** 60 s (NFX-12 §3). */ + readonly idleTimeoutMs: number; + /** 16 (NFX-12 §4). */ + readonly maxVideos: number; + /** 8 in free mode (NFX-12 §4). */ + readonly maxFreeRequestsPerChannel: number; +} + +/** + * Remote channel opens this connection rejects (an id that is no canonical address, a video + * not served, free mode's pay/1) before it closes: protomux 3.11 bounds neither pending nor + * rejected opens (`MAX_BACKLOG = Infinity`), and each costs a slot that is never reclaimed. + */ +export const MAX_REJECTED_OPENS = 64; + +export const NFX_LIMITS: NfxLimits = { + keepAliveMs: nfx.KEEPALIVE_INTERVAL_MS, + idleTimeoutMs: nfx.INACTIVITY_TIMEOUT_MS, + maxVideos: nfx.MAX_VIDEOS_PER_CONNECTION, + maxFreeRequestsPerChannel: nfx.MAX_FREE_REQUESTS_PER_CHANNEL, +}; + +/** What a channel needs from its connection (`files-server.ts`, `files-client.ts`, `pay-lane.ts`). */ +export interface ConnectionHost { + readonly mux: Protomux; + readonly pump: FilePump; + readonly limits: NfxLimits; + readonly remotePublicKey: Uint8Array; + /** Close the whole connection: a limit was exceeded, or the source is discarded. */ + destroy(reason: string): void; + /** Stop reading the stream while any holder holds (a sink is behind); idempotent per holder. */ + holdReads(holder: object): void; + releaseReads(holder: object): void; +} + +/** The seeder side of a connection: what it serves, and whether it takes pay/1 lanes. */ +export interface NfxServeOptions extends FilesServerOptions { + /** + * Accept a remote `nfx/pay/1` lane for a video the source serves (Phase 2's engine attaches + * here). Unset: free mode, and every pay/1 open is rejected. + */ + readonly onPay?: (video: nfx.VideoAddr) => PayLaneHandlers | undefined; +} + +export interface NfxConnectionOptions { + /** Set on the seeder side: the connection serves files. */ + readonly serve?: NfxServeOptions; + /** For the watcher side's sha256 (the platform library, `adapters/crypto.ts`). */ + readonly crypto?: SeederCrypto; + readonly limits?: Partial; + /** Told when the connection closed and why (a short reason). */ + readonly onClose?: (reason: string) => void; +} + +export class NfxConnection implements ConnectionHost { + readonly mux: Protomux; + readonly pump: FilePump; + readonly limits: NfxLimits; + /** The peer's Noise key; on the watcher side, the key it dialed (checked by `dialNfx`). */ + readonly remotePublicKey: Uint8Array; + private readonly videos = new Set(); + private readonly clients = new Map(); + private readonly holders = new Set(); + private rejectedOpens = 0; + private closeReason = 'closed'; + private isClosed = false; + + /** + * Wrap an OPENED secret-stream (from `dialNfx`, `listenNfx`, or an in-memory pair in tests). + * Throws if its handshake has not completed: the remote key is the account's identity. + */ + constructor( + readonly socket: NoiseSecretStream, + private readonly o: NfxConnectionOptions = {}, + ) { + if (socket.remotePublicKey === null) throw new Error('nfx: the stream is not open'); + this.remotePublicKey = socket.remotePublicKey; + this.limits = { ...NFX_LIMITS, ...o.limits }; + socket.setKeepAlive(this.limits.keepAliveMs); + socket.setTimeout(this.limits.idleTimeoutMs); + socket.on('error', (err) => { + this.closeReason = err.message; + }); + socket.on('close', () => { + this.isClosed = true; + this.pump.close(); + this.o.onClose?.(this.closeReason); + }); + this.mux = Protomux.from(socket); + this.pump = new FilePump(socket); + const serve = o.serve; + if (serve !== undefined) { + this.mux.pair({ protocol: nfx.FILES_PROTOCOL }, (id) => this.acceptFiles(id, serve)); + this.mux.pair({ protocol: nfx.PAY_PROTOCOL }, (id) => this.acceptPay(id, serve)); + } + } + + get closed(): boolean { + return this.isClosed || this.socket.destroyed; + } + + /** + * The watcher side of `nfx/files/1` for `video` on this connection (one per video), or + * `undefined` past the video limit, without `crypto`, or once the connection is closed. + */ + files(video: nfx.VideoAddr): FilesClient | undefined { + const key = video.toString(); + const known = this.clients.get(key); + if (known !== undefined && !known.closed) return known; + if (this.closed || this.o.crypto === undefined || !this.claimVideo(key)) return undefined; + try { + const client = new FilesClient(this, video, this.o.crypto); + this.clients.set(key, client); + return client; + } catch { + return undefined; + } + } + + /** The watcher side of `nfx/pay/1` for `video`, or `undefined` past the limit or once closed. */ + pay(video: nfx.VideoAddr, handlers: PayLaneHandlers): PayLane | undefined { + if (this.closed || !this.claimVideo(video.toString())) return undefined; + try { + return new PayLane(this, video, handlers); + } catch { + return undefined; + } + } + + destroy(reason: string): void { + if (this.closed) return; + this.closeReason = reason; + this.socket.destroy(); + } + + holdReads(holder: object): void { + this.holders.add(holder); + if (this.holders.size === 1) this.socket.pause(); + } + + releaseReads(holder: object): void { + if (!this.holders.delete(holder)) return; + if (this.holders.size === 0) this.socket.resume(); + } + + /** A video counted against the limit; `false` (and the seeder closes) once past it. */ + private claimVideo(key: string): boolean { + if (this.videos.has(key)) return true; + if (this.videos.size >= this.limits.maxVideos) return false; + this.videos.add(key); + return true; + } + + private async acceptFiles(id: Uint8Array | null, serve: NfxServeOptions): Promise { + const video = await this.acceptVideo(id, serve); + if (video === undefined || serveFiles(this, video, serve) === null) this.rejected(); + } + + private async acceptPay(id: Uint8Array | null, serve: NfxServeOptions): Promise { + const video = await this.acceptVideo(id, serve); + // Free mode (no `onPay`, or it declines): the open is rejected. + const handlers = video === undefined ? undefined : serve.onPay?.(video); + if (video === undefined || handlers === undefined) { + this.rejected(); + return; + } + try { + new PayLane(this, video, handlers); + } catch { + this.rejected(); // the stream went away while the handler ran + } + } + + /** Count a rejected remote open; past the budget the peer is closed. */ + private rejected(): void { + if (++this.rejectedOpens > MAX_REJECTED_OPENS) { + this.destroy('too many rejected channel opens'); + } + } + + /** + * The video a remote channel id names, if this connection may open it: a canonical address, + * served by the source, within the video limit. Otherwise `undefined`, and protomux rejects + * the open (or the connection is closed past the limit). + */ + private async acceptVideo( + id: Uint8Array | null, + serve: NfxServeOptions, + ): Promise { + if (id === null || id.byteLength > 256 || this.closed) return undefined; + let text: string; + try { + text = new TextDecoder('utf-8', { fatal: true }).decode(id); + } catch { + return undefined; + } + const video = nfx.VideoAddr.parse(text); + if (!video.ok || video.value.toString() !== text) return undefined; + if (!this.claimVideo(text)) { + this.destroy('more videos than one connection may carry'); + return undefined; + } + let served: boolean; + try { + served = await serve.source.hasVideo(video.value); + } catch { + served = false; + } + return served && !this.closed ? video.value : undefined; + } +} diff --git a/packages/seeder/src/nfx/endpoint.ts b/packages/seeder/src/nfx/endpoint.ts new file mode 100644 index 00000000..70f14931 --- /dev/null +++ b/packages/seeder/src/nfx/endpoint.ts @@ -0,0 +1,176 @@ +/** + * Dial and listen (NFX-12 §1, §3) on a HyperDHT node the caller owns: the worker's swarm DHT + * or one of its own, built with the network's bootstrap nodes from local configuration. This + * module only ever dials a key it is given; it never takes an address from received data. + * + * - The seeder listens on its per-network peer key pair; the watcher dials with its own + * per-network key pair (§1: never the same pair, never shared with another protocol). + * - The Noise IK handshake authenticates the dialed key; the dialer still closes a connection + * whose remote key is not the one it dialed (§3). + * - Local policy (§9: keys are free, so per-connection limits bound one connection, not one + * attacker): the listener caps connections in total and per remote key, in the DHT firewall + * before the handshake completes and again when a connection arrives. + */ +import type NoiseSecretStream from '@hyperswarm/secret-stream'; +import type { ConnectOptions, Server, ServerOptions } from 'hyperdht'; + +import type { SeederCrypto } from '../adapters/crypto.js'; +import { bytesEqual, toHex } from '../util/hex.js'; +import type { PeerKeyPair } from './bind.js'; +import { NfxConnection, type NfxLimits, type NfxServeOptions } from './connection.js'; + +/** The HyperDHT surface used here (`hyperdht` 6). */ +export interface NfxDht { + createServer(opts: ServerOptions, onconnection: (socket: NoiseSecretStream) => void): Server; + connect(remotePublicKey: Uint8Array, opts?: ConnectOptions): NoiseSecretStream; +} + +export const DEFAULT_MAX_CONNECTIONS = 64; +export const DEFAULT_MAX_CONNECTIONS_PER_PEER = 2; +export const DEFAULT_DIAL_TIMEOUT_MS = 30_000; + +export interface NfxListenOptions extends NfxServeOptions { + /** This network's peer key pair, the one its beacons' `hyperdht` endpoint names. */ + readonly keyPair: PeerKeyPair; + /** `true` REJECTS a dialer's Noise key before the handshake completes (a ban list). */ + readonly firewall?: (remotePublicKey: Uint8Array) => boolean; + readonly maxConnections?: number; + readonly maxConnectionsPerPeer?: number; + readonly limits?: Partial; + readonly onConnection?: (connection: NfxConnection) => void; +} + +export interface NfxListener { + readonly publicKey: Uint8Array; + readonly connections: number; + /** Stop listening and close every connection. */ + close(): Promise; +} + +export async function listenNfx(dht: NfxDht, o: NfxListenOptions): Promise { + const maxConnections = o.maxConnections ?? DEFAULT_MAX_CONNECTIONS; + const maxPerPeer = o.maxConnectionsPerPeer ?? DEFAULT_MAX_CONNECTIONS_PER_PEER; + const live = new Set(); + const perPeer = new Map(); + let closing = false; + + const full = (key: string): boolean => + live.size >= maxConnections || (perPeer.get(key) ?? 0) >= maxPerPeer; + + const accept = (socket: NoiseSecretStream): void => { + socket.on('error', () => undefined); // NfxConnection records why; this one is for early exits + const remote = socket.remotePublicKey; + const key = remote === null ? '' : toHex(remote); + // Re-checked: the firewall ran before the handshake finished, and others may have come since. + if (closing || remote === null || full(key)) { + socket.destroy(); + return; + } + const release = (): void => { + const n = (perPeer.get(key) ?? 1) - 1; + if (n <= 0) perPeer.delete(key); + else perPeer.set(key, n); + }; + let connection: NfxConnection; + try { + connection = new NfxConnection(socket, { + serve: o, + ...(o.limits !== undefined ? { limits: o.limits } : {}), + onClose: () => { + if (live.delete(connection)) release(); + }, + }); + } catch { + socket.destroy(); + return; + } + live.add(connection); + perPeer.set(key, (perPeer.get(key) ?? 0) + 1); + o.onConnection?.(connection); + }; + + const server = dht.createServer( + { + firewall: (remotePublicKey) => + closing || o.firewall?.(remotePublicKey) === true || full(toHex(remotePublicKey)), + }, + accept, + ); + await server.listen(o.keyPair); + + return { + publicKey: o.keyPair.publicKey, + get connections(): number { + return live.size; + }, + async close(): Promise { + closing = true; + for (const c of [...live]) c.destroy('the listener closed'); + await server.close(); + }, + }; +} + +export interface NfxDialOptions { + /** This network's watcher key pair: its public key is the NFX-07 §3 peer of the account. */ + readonly keyPair: PeerKeyPair; + /** sha256 for fetched files (the platform library). */ + readonly crypto: SeederCrypto; + readonly limits?: Partial; + /** Give up if the connection is not open by then. */ + readonly timeoutMs?: number; + readonly onClose?: (reason: string) => void; +} + +export type DialResult = + | { readonly ok: true; readonly connection: NfxConnection } + | { + readonly ok: false; + readonly reason: 'invalid-argument' | 'unreachable' | 'timeout' | 'wrong-key'; + }; + +/** Dial `peer` (a 32-byte peer key from a valid beacon, NFX-12 §3). Never throws. */ +export async function dialNfx( + dht: NfxDht, + peer: Uint8Array, + o: NfxDialOptions, +): Promise { + if (peer.byteLength !== 32) return { ok: false, reason: 'invalid-argument' }; + let socket: NoiseSecretStream; + try { + socket = dht.connect(peer, { keyPair: o.keyPair }); + } catch { + return { ok: false, reason: 'unreachable' }; + } + socket.on('error', () => undefined); // a refusal or timeout surfaces as `opened === false` + + let timer: ReturnType | undefined; + const deadline = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => { + resolve('timeout'); + }, o.timeoutMs ?? DEFAULT_DIAL_TIMEOUT_MS); + }); + const outcome = await Promise.race([socket.opened, deadline]); + clearTimeout(timer); + + if (outcome !== true) { + socket.destroy(); + return { ok: false, reason: outcome === 'timeout' ? 'timeout' : 'unreachable' }; + } + const remote = socket.remotePublicKey; + if (remote === null || !bytesEqual(remote, peer)) { + socket.destroy(); + return { ok: false, reason: 'wrong-key' }; + } + try { + const connection = new NfxConnection(socket, { + crypto: o.crypto, + ...(o.limits !== undefined ? { limits: o.limits } : {}), + ...(o.onClose !== undefined ? { onClose: o.onClose } : {}), + }); + return { ok: true, connection }; + } catch { + socket.destroy(); + return { ok: false, reason: 'unreachable' }; + } +} diff --git a/packages/seeder/src/nfx/files-client.ts b/packages/seeder/src/nfx/files-client.ts new file mode 100644 index 00000000..4c4a714c --- /dev/null +++ b/packages/seeder/src/nfx/files-client.ts @@ -0,0 +1,333 @@ +/** + * The watcher side of `nfx/files/1` (NFX-12 §4) on one channel: whole files by sha256, the + * receiver duties of §4 and the integrity rule of §6. + * + * - Every `data` payload is copied out of the receive buffer (the codec hands out a view) and + * written to the caller's sink in order, one write at a time. While more than 1 MiB waits + * for the sink the connection stops reading, so a slow sink cannot grow memory unboundedly. + * - A file is capped at its size: a byte past it, or `end` before it, is a mismatch. Its + * sha256 is checked before the result says `ok`, and a mismatch drops the whole connection: + * the source is discarded for the session (NFX-05 §4). The sink's bytes are the caller's to + * discard on any failure; only `ok` means they are the file. + * - `data`, `end` or `missing` for a `req` that is not open, a seeder's `get`, or a trapped + * unknown type close the channel; its open requests fail `protocol`. + * - A request that gets no frame for the idle limit (60 s) fails `timeout`, and the channel is + * closed to abandon it: a seeder's keepalives hold the connection open, so the connection's + * own idle rule cannot catch a transfer that simply stops. + */ +import { nfx } from '@sovit/core'; +import type { ProtomuxChannel, ProtomuxMessage } from 'protomux'; + +import type { SeederCrypto, Sha256Hasher } from '../adapters/crypto.js'; +import { fromHex } from '../util/hex.js'; +import type { ConnectionHost } from './connection.js'; +import { addTraps } from './traps.js'; + +/** Where a fetched file's bytes go, in order (a spool file, normally). */ +export interface NfxSink { + write(chunk: Uint8Array): void | Promise; +} + +export interface FetchOptions { + /** The file's sha256, 64 lowercase hex. */ + readonly sha: string; + /** + * Its `files[].size`. With `exact: false`, a cap instead: the hash list, fetched as `sha` = + * the manifest `root`, is any length up to 16 MiB. + */ + readonly size: number; + readonly exact?: boolean; + readonly sink: NfxSink; + /** Fail `timeout` if the file is not complete by then. Unset: only the 60 s stall rule. */ + readonly timeoutMs?: number; +} + +export type FetchFailure = + /** The seeder does not serve it (free mode). */ + | 'missing' + /** The bytes' sha256 is not `sha`: the connection was dropped. */ + | 'mismatch' + /** More bytes than `size`: the connection was dropped. */ + | 'too-long' + /** `end` before `size` bytes: the connection was dropped. */ + | 'too-short' + /** The seeder broke `nfx/files/1`; the channel was closed. */ + | 'protocol' + /** The seeder does not serve this video on this connection. */ + | 'unserved' + | 'closed' + | 'timeout' + /** As many requests are open on this channel as NFX-12 §4 allows. */ + | 'busy' + /** A request for the same `sha` is open on this channel. */ + | 'duplicate' + | 'invalid-argument' + /** The sink threw: the channel was closed, since a request cannot be cancelled. */ + | 'sink'; + +export type FetchResult = + | { readonly ok: true; readonly bytes: number } + | { readonly ok: false; readonly reason: FetchFailure }; + +/** Bytes waiting for the sink above which the connection stops reading, and below which it resumes. */ +const SINK_HIGH_BYTES = 1024 * 1024; +const SINK_LOW_BYTES = 256 * 1024; + +interface Transfer { + readonly sha: string; + readonly size: number; + readonly exact: boolean; + readonly sink: NfxSink; + readonly hasher: Sha256Hasher; + received: number; + waiting: number; + /** Restart the stall timer: a frame of this request arrived. */ + alive(): void; + writes: Promise; + sinkFailed: boolean; + settle(result: FetchResult): void; +} + +export class FilesClient { + private readonly channel: ProtomuxChannel; + private readonly get: ProtomuxMessage; + private readonly opened: Promise; + private readonly open = new Map(); + private nextReq = 1; + private everOpened = false; + + /** Use `NfxConnection.files(video)`. Throws if the mux will not open the channel. */ + constructor( + private readonly host: ConnectionHost, + readonly video: nfx.VideoAddr, + private readonly crypto: SeederCrypto, + ) { + const ch = host.mux.createChannel({ + protocol: nfx.FILES_PROTOCOL, + id: video.channelId(), + onclose: () => { + this.failAll(this.everOpened ? 'closed' : 'unserved'); + }, + }); + if (ch === null) { + throw new Error('nfx/files/1: the channel is already open or the stream is gone'); + } + this.channel = ch; + // Pinned order (NFX-12 §4): 0 get, 1 data, 2 end, 3 missing. + this.get = ch.addMessage({ + onmessage: () => { + this.protocolError(); + }, + }); + ch.addMessage({ + onmessage: (bytes) => { + this.onData(bytes); + }, + }); + ch.addMessage({ + onmessage: (bytes) => { + this.onEnd(bytes); + }, + }); + ch.addMessage({ + onmessage: (bytes) => { + this.onMissing(bytes); + }, + }); + addTraps(ch, () => { + this.protocolError(); + }); + ch.open(); + this.opened = ch.fullyOpened().then((opened) => { + this.everOpened = opened; + return opened; + }); + } + + get openRequests(): number { + return this.open.size; + } + + get closed(): boolean { + return this.channel.closed; + } + + /** Fetch one whole file. Resolves once; never throws. */ + async fetch(o: FetchOptions): Promise { + const exact = o.exact ?? true; + if ( + !/^[0-9a-f]{64}$/.test(o.sha) || + !Number.isSafeInteger(o.size) || + o.size < 0 || + (!exact && o.size > nfx.MAX_HASH_LIST_BYTES) + ) { + return { ok: false, reason: 'invalid-argument' }; + } + if (this.open.size >= this.host.limits.maxFreeRequestsPerChannel) { + return { ok: false, reason: 'busy' }; + } + for (const t of this.open.values()) + if (t.sha === o.sha) return { ok: false, reason: 'duplicate' }; + if (!(await this.opened) || this.channel.closed) { + return { ok: false, reason: this.everOpened ? 'closed' : 'unserved' }; + } + // Re-checked: other fetches may have started while this one waited for the channel. + if (this.open.size >= this.host.limits.maxFreeRequestsPerChannel) { + return { ok: false, reason: 'busy' }; + } + for (const t of this.open.values()) + if (t.sha === o.sha) return { ok: false, reason: 'duplicate' }; + const req = this.allocateReq(); + const frame = nfx.encodeFilesMessage({ type: 'get', req, sha: fromHex(o.sha) }); + if (!frame.ok) return { ok: false, reason: 'invalid-argument' }; + + return new Promise((resolve) => { + let timer: ReturnType | undefined; + let stall: ReturnType | undefined; + const abandon = (): void => { + // No cancel in nfx/files/1 (§4): abandoning a transfer closes the channel. + transfer.settle({ ok: false, reason: 'timeout' }); + this.channel.close(); + }; + const transfer: Transfer = { + sha: o.sha, + size: o.size, + exact, + sink: o.sink, + hasher: this.crypto.createSha256(), + received: 0, + waiting: 0, + alive: () => { + if (stall !== undefined) clearTimeout(stall); + stall = setTimeout(() => { + // Bytes still waiting for the sink: the silence is ours (reads are held), not the seeder's. + if (transfer.waiting > 0) transfer.alive(); + else abandon(); + }, this.host.limits.idleTimeoutMs); + }, + writes: Promise.resolve(), + sinkFailed: false, + settle: (result) => { + if (this.open.get(req) !== transfer) return; + this.open.delete(req); + if (timer !== undefined) clearTimeout(timer); + if (stall !== undefined) clearTimeout(stall); + this.host.releaseReads(transfer); + resolve(result); + }, + }; + this.open.set(req, transfer); + transfer.alive(); + if (o.timeoutMs !== undefined) timer = setTimeout(abandon, o.timeoutMs); + this.get.send(frame.value.bytes); + }); + } + + /** Close the channel; open requests fail `closed`. */ + close(): void { + this.channel.close(); + } + + private onData(bytes: Uint8Array): void { + const m = this.decode(1, bytes); + if (m?.type !== 'data') return; + const t = this.open.get(m.req); + if (t === undefined) { + this.protocolError(); + return; + } + if (t.received + m.payload.byteLength > t.size) { + this.discardSource(t, 'too-long'); + return; + } + t.alive(); + const chunk = m.payload.slice(); + t.received += chunk.byteLength; + t.hasher.update(chunk); + t.waiting += chunk.byteLength; + if (t.waiting > SINK_HIGH_BYTES) this.host.holdReads(t); + t.writes = t.writes.then(async () => { + if (t.sinkFailed) return; + try { + await t.sink.write(chunk); + } catch { + t.sinkFailed = true; + t.settle({ ok: false, reason: 'sink' }); + this.channel.close(); + return; + } + t.waiting -= chunk.byteLength; + if (t.waiting < SINK_LOW_BYTES) this.host.releaseReads(t); + }); + } + + private onEnd(bytes: Uint8Array): void { + const m = this.decode(2, bytes); + if (m?.type !== 'end') return; + const t = this.open.get(m.req); + if (t === undefined) { + this.protocolError(); + return; + } + if (t.exact && t.received !== t.size) { + this.discardSource(t, 'too-short'); + return; + } + void t.writes.then(() => { + if (t.sinkFailed) return; + if (t.hasher.digestHex() !== t.sha) { + this.discardSource(t, 'mismatch'); + return; + } + t.settle({ ok: true, bytes: t.received }); + }); + } + + private onMissing(bytes: Uint8Array): void { + const m = this.decode(3, bytes); + if (m?.type !== 'missing') return; + const t = this.open.get(m.req); + if (t === undefined) { + this.protocolError(); + return; + } + // A refusal answers a `get`; after bytes of the same request it is a protocol error. + if (t.received > 0) { + t.settle({ ok: false, reason: 'protocol' }); + this.protocolError(); + return; + } + t.settle({ ok: false, reason: 'missing' }); + } + + private decode(index: number, bytes: Uint8Array): nfx.FilesMessage | undefined { + const m = nfx.decodeFilesMessage(index, bytes); + if (m.ok) return m.value; + this.protocolError(); + return undefined; + } + + /** NFX-05 §4: a source that served a mismatch is discarded for the session. */ + private discardSource(t: Transfer, reason: 'too-long' | 'too-short' | 'mismatch'): void { + t.settle({ ok: false, reason }); + this.host.destroy(`the seeder served a ${reason} file`); + } + + private protocolError(): void { + this.failAll('protocol'); + this.channel.close(); + } + + private failAll(reason: FetchFailure): void { + for (const t of [...this.open.values()]) t.settle({ ok: false, reason }); + } + + /** A u32 not in use on this channel (NFX-12 §4: unique among the open requests). */ + private allocateReq(): number { + for (;;) { + const req = this.nextReq; + this.nextReq = req === 0xffff_ffff ? 0 : req + 1; + if (!this.open.has(req)) return req; + } + } +} diff --git a/packages/seeder/src/nfx/files-server.ts b/packages/seeder/src/nfx/files-server.ts new file mode 100644 index 00000000..59fe8cda --- /dev/null +++ b/packages/seeder/src/nfx/files-server.ts @@ -0,0 +1,187 @@ +/** + * The seeder side of `nfx/files/1` (NFX-12 §4) on one channel, i.e. one video on one + * connection. Free mode only: a request the node will not or cannot serve is answered + * `missing`. Paid mode (Phase 2) plugs in at `NfxAdmission`, where a pay/1 session decides. + * + * - Only a file the source lists for THIS video (its verified hash list, or the hash list + * itself as `sha` = the manifest `root`) is served, so a channel is never an oracle for the + * rest of the node's store. + * - Bytes go out in `data` frames of at most 64 KiB, then `end`, every frame through the + * connection's pump (flow control, `pump.ts`). Each frame is read inside the pump's turn, so + * a connection holds one chunk in memory however many requests it has open (a peer could + * otherwise make the node buffer 16 videos × 8 requests × 64 KiB per connection); the cost + * is that a slow read delays the other transfers on the same connection only. + * - A malformed message, a message a seeder never receives (`data`, `end`, `missing`), an + * unknown type in the trapped range, or a `req` already open closes the channel. More open + * requests than the limit closes the connection (§4 "a peer that exceeds one is closed"). + */ +import { nfx } from '@sovit/core'; +import type { ProtomuxChannel, ProtomuxMessage } from 'protomux'; + +import { toHex } from '../util/hex.js'; +import type { ConnectionHost } from './connection.js'; +import { addTraps } from './traps.js'; + +/** One file a node serves on a channel. */ +export interface NfxServedFile { + /** The file's byte length (its `files[].size`, or the hash list's length). */ + readonly size: number; + /** Exactly `length` bytes at `offset`; anything else fails the transfer. */ + read(offset: number, length: number): Promise; +} + +/** What a node serves: its stores, as the transport sees them. */ +export interface NfxFileSource { + /** Whether this node serves `video` at all. A channel for any other video is rejected. */ + hasVideo(video: nfx.VideoAddr): boolean | Promise; + /** + * File `sha` (64 lowercase hex) of `video`: only a file listed in that video's hash list + * this node holds and verified (NFX-05 §5), or the hash list itself when `sha` is the + * manifest `root`. `undefined` for anything else, even a file held for another video. + */ + file( + video: nfx.VideoAddr, + sha: string, + ): NfxServedFile | undefined | Promise; +} + +/** A request as the admission hook sees it. */ +export interface NfxRequest { + readonly video: nfx.VideoAddr; + readonly sha: string; + /** The watcher's Noise key: the NFX-07 §3 peer of its account on this transport. */ + readonly remotePublicKey: Uint8Array; +} + +/** + * Where paid mode decides (Phase 2). Free mode admits every request the source serves; a + * `false` here is answered `missing`. + */ +export type NfxAdmission = (request: NfxRequest) => boolean | Promise; + +export interface FilesServerOptions { + readonly source: NfxFileSource; + readonly admission?: NfxAdmission; + /** Told why the channel or connection was closed (a short reason, never file bytes). */ + readonly onProtocolError?: (why: string) => void; +} + +/** Open the seeder side of `nfx/files/1` for `video` on `host`; `null` if the mux refused it. */ +export function serveFiles( + host: ConnectionHost, + video: nfx.VideoAddr, + o: FilesServerOptions, +): ProtomuxChannel | null { + const ch = host.mux.createChannel({ protocol: nfx.FILES_PROTOCOL, id: video.channelId() }); + if (ch === null) return null; + const open = new Set(); + const fail = (why: string): void => { + o.onProtocolError?.(why); + ch.close(); + }; + + // Pinned order (NFX-12 §4): 0 get, 1 data, 2 end, 3 missing. + ch.addMessage({ + onmessage: (bytes) => { + onGet(bytes); + }, + }); + const data = ch.addMessage({ + onmessage: () => { + fail('a seeder does not receive `data`'); + }, + }); + const end = ch.addMessage({ + onmessage: () => { + fail('a seeder does not receive `end`'); + }, + }); + const missing = ch.addMessage({ + onmessage: () => { + fail('a seeder does not receive `missing`'); + }, + }); + addTraps(ch, () => { + fail('unknown nfx/files/1 message type'); + }); + ch.open(); + return ch; + + function onGet(bytes: Uint8Array): void { + const decoded = nfx.decodeFilesMessage(0, bytes); + if (!decoded.ok || decoded.value.type !== 'get') { + fail(decoded.ok ? 'not a `get`' : decoded.error.reason); + return; + } + const { req, sha } = decoded.value; + if (open.has(req)) { + fail('a `req` that is already open'); + return; + } + if (open.size >= host.limits.maxFreeRequestsPerChannel) { + host.destroy('too many open requests on one channel'); + return; + } + open.add(req); + void serve(req, toHex(sha)).finally(() => { + open.delete(req); + }); + } + + async function serve(req: number, sha: string): Promise { + let file: NfxServedFile | undefined; + try { + const request = { video, sha, remotePublicKey: host.remotePublicKey }; + const admitted = o.admission === undefined ? true : await o.admission(request); + file = admitted ? await o.source.file(video, sha) : undefined; + } catch { + fail('the source failed'); + return; + } + if (ch.closed) return; + if (file === undefined) { + send(missing, { type: 'missing', req }); + return; + } + const size = file.size; + if (!Number.isSafeInteger(size) || size < 0) { + fail('the source gave a bad size'); + return; + } + for (let offset = 0; offset < size; offset += nfx.MAX_DATA_PAYLOAD) { + const length = Math.min(nfx.MAX_DATA_PAYLOAD, size - offset); + if (!(await host.pump.turn())) return; + let chunk: Uint8Array; + try { + chunk = await file.read(offset, length); + } catch { + host.pump.skip(); + fail('the source failed mid-file'); + return; + } + if (ch.closed) { + host.pump.skip(); + return; + } + if (chunk.byteLength !== length) { + host.pump.skip(); + fail('the source returned a short or long read'); + return; + } + const frame = nfx.encodeFilesMessage({ type: 'data', req, payload: chunk }); + if (!frame.ok) { + host.pump.skip(); + fail(frame.error.reason); + return; + } + host.pump.sent(frame.value.bytes.byteLength, data.send(frame.value.bytes)); + } + // After the last `data` frame on the same ordered stream, so it cannot overtake it. + send(end, { type: 'end', req }); + } + + function send(message: ProtomuxMessage, m: nfx.FilesMessage): void { + const frame = nfx.encodeFilesMessage(m); + if (frame.ok) message.send(frame.value.bytes); + } +} diff --git a/packages/seeder/src/nfx/index.ts b/packages/seeder/src/nfx/index.ts new file mode 100644 index 00000000..949c47ed --- /dev/null +++ b/packages/seeder/src/nfx/index.ts @@ -0,0 +1,43 @@ +/** + * The NFX-12 HyperDHT transport (nfx `spec/nfx-12-hyperdht.md`, ADR 0020): dial a seeder by + * its peer key, fetch whole files over `nfx/files/1`, carry NFX-07 over `nfx/pay/1`. Free mode + * only; the wire layouts are `@sovit/core`'s `nfx` codec. It runs beside the legacy Hypercore + * path and shares neither keys nor stores with it. + */ +export { hyperdhtBind, hyperdhtEndpoint } from './bind.js'; +export type { HyperdhtEndpoint, PeerKeyPair } from './bind.js'; +export { MAX_REJECTED_OPENS, NFX_LIMITS, NfxConnection } from './connection.js'; +export type { + ConnectionHost, + NfxConnectionOptions, + NfxLimits, + NfxServeOptions, +} from './connection.js'; +export { + DEFAULT_DIAL_TIMEOUT_MS, + DEFAULT_MAX_CONNECTIONS, + DEFAULT_MAX_CONNECTIONS_PER_PEER, + dialNfx, + listenNfx, +} from './endpoint.js'; +export type { + DialResult, + NfxDht, + NfxDialOptions, + NfxListener, + NfxListenOptions, +} from './endpoint.js'; +export { FilesClient } from './files-client.js'; +export type { FetchFailure, FetchOptions, FetchResult, NfxSink } from './files-client.js'; +export type { + FilesServerOptions, + NfxAdmission, + NfxFileSource, + NfxRequest, + NfxServedFile, +} from './files-server.js'; +export { PayLane } from './pay-lane.js'; +export type { PayLaneHandlers } from './pay-lane.js'; +export { FilePump } from './pump.js'; +export type { DrainSource, PumpStats } from './pump.js'; +export { TRAPPED_TYPES } from './traps.js'; diff --git a/packages/seeder/src/nfx/pay-lane.ts b/packages/seeder/src/nfx/pay-lane.ts new file mode 100644 index 00000000..7c83fee0 --- /dev/null +++ b/packages/seeder/src/nfx/pay-lane.ts @@ -0,0 +1,96 @@ +/** + * `nfx/pay/1` (NFX-12 §5) on one channel: the lane NFX-07 messages ride, and nothing more. + * There is no payment logic here (Phase 2's engine owns sessions, quotes and tokens); the lane + * only holds the codec's rules both ways (UTF-8, no newline, at most 32,767 bytes) and closes + * on a message that breaks them or on a trapped unknown type. + * + * Its messages are written straight to the stream, never through the files pump, so a pay/1 + * message waits behind at most one `data` frame on a buffer below the stream's high-water + * mark (`pump.ts`), which keeps NFX-12 §3's 60 s rule on any working link. + */ +import { nfx } from '@sovit/core'; +import type { ProtomuxChannel, ProtomuxMessage } from 'protomux'; + +import type { ConnectionHost } from './connection.js'; +import { addTraps } from './traps.js'; + +export interface PayLaneHandlers { + /** One NFX-07 §2 message as text, for the engine to parse; reply on `lane`. */ + readonly onMessage: (text: string, lane: PayLane) => void; + /** The lane closed: `remote` when the peer closed or rejected it. */ + readonly onClose?: (remote: boolean) => void; + /** Told why the lane closed itself (a short reason, never message content). */ + readonly onProtocolError?: (why: string) => void; +} + +export class PayLane { + private readonly channel: ProtomuxChannel; + private readonly msg: ProtomuxMessage; + + /** Use `NfxConnection.pay(video, …)` or the listener's `onPay`. Throws if the mux will not open it. */ + constructor( + host: ConnectionHost, + readonly video: nfx.VideoAddr, + private readonly h: PayLaneHandlers, + ) { + const ch = host.mux.createChannel({ + protocol: nfx.PAY_PROTOCOL, + id: video.channelId(), + onclose: (remote) => { + this.h.onClose?.(remote); + }, + }); + if (ch === null) { + throw new Error('nfx/pay/1: the channel is already open or the stream is gone'); + } + this.channel = ch; + // Pinned (NFX-12 §5): 0 msg. + this.msg = ch.addMessage({ + onmessage: (bytes) => { + this.onMessage(bytes); + }, + }); + addTraps(ch, () => { + this.fail('unknown nfx/pay/1 message type'); + }); + ch.open(); + } + + get closed(): boolean { + return this.channel.closed; + } + + /** `true` once the peer opened its side, `false` if it rejected the lane or it closed. */ + opened(): Promise { + return this.channel.fullyOpened(); + } + + /** + * Send one message. `ok: false` if it breaks the codec's rules (nothing was sent); + * `ok: true` carries the stream's `write()` result. + */ + send(text: string): nfx.NfxResult { + const bytes = nfx.encodePayMessage(text); + if (!bytes.ok) return bytes; + if (this.channel.closed) return { ok: true, value: false }; + return { ok: true, value: this.msg.send(bytes.value) }; + } + + close(): void { + this.channel.close(); + } + + private onMessage(bytes: Uint8Array): void { + const text = nfx.decodePayMessage(0, bytes); + if (!text.ok) { + this.fail(text.error.reason); + return; + } + this.h.onMessage(text.value, this); + } + + private fail(why: string): void { + this.h.onProtocolError?.(why); + this.channel.close(); + } +} diff --git a/packages/seeder/src/nfx/pump.ts b/packages/seeder/src/nfx/pump.ts new file mode 100644 index 00000000..3e153c9b --- /dev/null +++ b/packages/seeder/src/nfx/pump.ts @@ -0,0 +1,132 @@ +/** + * NFX-12 §4 flow control for one connection: every `nfx/files/1` frame of every channel on the + * connection goes through one pump, one frame at a time. A frame is written only while the + * stream is not asking for a drain; after a write that returns `false`, nothing more is written + * until `'drain'` (streamx: the buffer has emptied). So at most one frame is ever written on top + * of a buffer below the stream's high-water mark (16 KiB): ≤ 16 KiB + 64 KiB + 4 B ahead of the + * transport, inside the 256 KiB cap, and a pay/1 message (written directly, never through the + * pump) waits behind no more than that. + * + * Waiting transfers are served in arrival order, so concurrent requests interleave frame by + * frame instead of one file starving the rest. + */ +import { nfx } from '@sovit/core'; + +/** What the pump needs from the connection's stream. */ +export interface DrainSource { + on(event: 'drain' | 'close', cb: () => void): unknown; + off(event: 'drain' | 'close', cb: () => void): unknown; +} + +/** One frame's worth of the cap: 64 KiB of payload plus the 4-byte `req`. */ +const MAX_FRAME_BYTES = nfx.MAX_DATA_PAYLOAD + 4; +/** streamx's default high-water mark; secret-stream does not change it. */ +const STREAM_HIGH_WATER_MARK = 16_384; + +// The bound in the header holds by construction; keep it true if either constant moves. +if (STREAM_HIGH_WATER_MARK + MAX_FRAME_BYTES > nfx.MAX_QUEUED_FILE_BYTES) { + throw new Error('nfx pump: a frame on a full buffer would exceed the 256 KiB queue cap'); +} + +export interface PumpStats { + /** Frames written. */ + readonly frames: number; + /** Bytes written. */ + readonly bytes: number; + /** Writes that left the stream asking for a drain. */ + readonly stalls: number; + /** + * The most bytes that can have been queued ahead of the transport at once, as an upper + * bound: a write that returned `true` leaves under 16 KiB buffered, a drain leaves none. + */ + readonly maxQueuedBytes: number; +} + +export class FilePump { + private blocked = false; + private closed = false; + /** One turn is out at a time: the holder writes, then `sent` or `skip` hands it on. */ + private turnTaken = false; + private readonly waiters: ((open: boolean) => void)[] = []; + private frames = 0; + private bytes = 0; + private stalls = 0; + /** Upper bound on the bytes buffered in the stream now. */ + private queued = 0; + private maxQueued = 0; + + private readonly onDrain = (): void => { + this.blocked = false; + this.queued = 0; + this.release(); + }; + + private readonly onClose = (): void => { + this.close(); + }; + + constructor(private readonly stream: DrainSource) { + stream.on('drain', this.onDrain); + stream.on('close', this.onClose); + } + + /** + * Resolves `true` when the caller may write one frame now (synchronously, then `sent`), or + * `false` once the connection is closed: the caller stops. + */ + turn(): Promise { + if (this.closed) return Promise.resolve(false); + return new Promise((resolve) => { + this.waiters.push(resolve); + this.release(); + }); + } + + /** Record a frame of `byteLength` written in this turn and whether the stream took it without a drain. */ + sent(byteLength: number, drained: boolean): void { + this.frames++; + this.bytes += byteLength; + this.queued += byteLength; + if (drained) { + // streamx accepted it below its high-water mark: less than that is buffered now. + this.queued = Math.min(this.queued, STREAM_HIGH_WATER_MARK - 1); + } else { + this.stalls++; + this.blocked = true; + } + if (this.queued > this.maxQueued) this.maxQueued = this.queued; + this.turnTaken = false; + this.release(); + } + + /** Give the turn back without writing (the transfer ended or failed). */ + skip(): void { + this.turnTaken = false; + this.release(); + } + + get stats(): PumpStats { + return { + frames: this.frames, + bytes: this.bytes, + stalls: this.stalls, + maxQueuedBytes: this.maxQueued, + }; + } + + close(): void { + if (this.closed) return; + this.closed = true; + this.stream.off('drain', this.onDrain); + this.stream.off('close', this.onClose); + for (const w of this.waiters.splice(0)) w(false); + } + + private release(): void { + if (this.closed || this.blocked || this.turnTaken) return; + const next = this.waiters.shift(); + if (next === undefined) return; + this.turnTaken = true; + next(true); + } +} diff --git a/packages/seeder/src/nfx/traps.ts b/packages/seeder/src/nfx/traps.ts new file mode 100644 index 00000000..65f5259b --- /dev/null +++ b/packages/seeder/src/nfx/traps.ts @@ -0,0 +1,15 @@ +/** + * Protomux numbers a channel's message types in `addMessage` order and silently drops a type + * the channel did not add (protomux 3.11.0 `Channel._recv`). NFX-12 §3–§4 make an unknown + * type a protocol error, so after its pinned types each channel adds a few trap types that + * close it. Indices past the traps are still dropped unseen: harmless, since a dropped + * message changes no state, but not the close the spec asks for. + */ +import type { ProtomuxChannel } from 'protomux'; + +/** Trap types added after a protocol's pinned ones. */ +export const TRAPPED_TYPES = 4; + +export function addTraps(channel: ProtomuxChannel, onTrap: () => void): void { + for (let i = 0; i < TRAPPED_TYPES; i++) channel.addMessage({ onmessage: onTrap }); +} diff --git a/packages/seeder/src/portable.ts b/packages/seeder/src/portable.ts index 94ccb972..acb8615b 100644 --- a/packages/seeder/src/portable.ts +++ b/packages/seeder/src/portable.ts @@ -68,6 +68,49 @@ export { } from './store/core-policies.js'; export type { PersistedBan } from './store/ban-list.js'; +// NFX-12 transport (HyperDHT + Protomux; ADR 0020): beside the legacy path, sharing nothing +export { + DEFAULT_DIAL_TIMEOUT_MS, + DEFAULT_MAX_CONNECTIONS, + DEFAULT_MAX_CONNECTIONS_PER_PEER, + FilePump, + FilesClient, + MAX_REJECTED_OPENS, + NFX_LIMITS, + NfxConnection, + PayLane, + TRAPPED_TYPES, + dialNfx, + hyperdhtBind, + hyperdhtEndpoint, + listenNfx, +} from './nfx/index.js'; +export type { + ConnectionHost, + DialResult, + DrainSource, + FetchFailure, + FetchOptions, + FetchResult, + FilesServerOptions, + HyperdhtEndpoint, + NfxAdmission, + NfxConnectionOptions, + NfxDht, + NfxDialOptions, + NfxFileSource, + NfxLimits, + NfxListener, + NfxListenOptions, + NfxRequest, + NfxServedFile, + NfxServeOptions, + NfxSink, + PayLaneHandlers, + PeerKeyPair, + PumpStats, +} from './nfx/index.js'; + // Payment export { FlushScheduler } from './payment/flush-scheduler.js'; export type { FlushResult, FlushTrigger } from './payment/flush-scheduler.js'; From 6d6964f1437a2706105636dfac7d8de3ce9b3123 Mon Sep 17 00:00:00 2001 From: sovtech Date: Thu, 22 Oct 2026 17:35:57 -0600 Subject: [PATCH 3/5] test(seeder/nfx): the NFX-12 transport in memory, over a local testnet, and its bind vector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - bind: reproduces `beacon-hyperdht.json`'s signature byte for byte from its throwaway peer seed, verifies with the codec only for its own seeder and network, refuses bad arguments. - pump: against a stand-in with streamx's write/drain semantics — never a write on a buffer that asked for a drain, the real queue at most one frame over 16 KiB, transfers interleaved in arrival order, waiters answered `false` and listeners removed on close. - transport (a real secret-stream pair joined in memory): multi-frame and zero-byte files byte for byte; the hash list against a cap; `missing` for a file held only for another video; `unserved`; poison, too-long and too-short each drop the source; bad arguments, a duplicate sha and a ninth request refused on the watcher side; a slow sink with reads held and the seeder's queue within 256 KiB; a stalled transfer abandoned while keepalives hold the connection; a throwing sink; keepalives and the idle close (short intervals); the pay/1 lane both ways, refused in free mode, closed on a newline. Hostile watchers on raw Protomux: a trapped unknown type and a reused `req` close the channel; a ninth request, a 17th video and 65 rejected opens close the connection; a non-canonical channel id is rejected. - endpoint (a local hyperdht testnet, as the swarm tests use): dial by key and fetch a multi-frame file, the seeder seeing the watcher's key; a firewalled key and a second connection from one key refused; a peer key that is not 32 bytes. --- .../seeder/src/nfx/__tests__/bind.test.ts | 80 +++ .../__tests__/endpoint.integration.test.ts | 141 ++++++ packages/seeder/src/nfx/__tests__/helpers.ts | 148 ++++++ .../seeder/src/nfx/__tests__/pump.test.ts | 106 ++++ .../src/nfx/__tests__/transport.test.ts | 460 ++++++++++++++++++ 5 files changed, 935 insertions(+) create mode 100644 packages/seeder/src/nfx/__tests__/bind.test.ts create mode 100644 packages/seeder/src/nfx/__tests__/endpoint.integration.test.ts create mode 100644 packages/seeder/src/nfx/__tests__/helpers.ts create mode 100644 packages/seeder/src/nfx/__tests__/pump.test.ts create mode 100644 packages/seeder/src/nfx/__tests__/transport.test.ts diff --git a/packages/seeder/src/nfx/__tests__/bind.test.ts b/packages/seeder/src/nfx/__tests__/bind.test.ts new file mode 100644 index 00000000..4581ad0b --- /dev/null +++ b/packages/seeder/src/nfx/__tests__/bind.test.ts @@ -0,0 +1,80 @@ +/** + * `bind` (NFX-12 §2) made here verifies with the codec's verifier, and reproduces the vendored + * vector's signature byte for byte (Ed25519 is deterministic; the vector's throwaway peer + * seed is the key pair `hyperdht` derives from it). + */ +import { readFileSync } from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import { nfx } from '@sovit/core'; +import DHT from 'hyperdht'; +import { describe, expect, it } from 'vitest'; + +import { fromHex, toHex } from '../../util/hex.js'; +import { hyperdhtBind, hyperdhtEndpoint } from '../bind.js'; + +const VECTORS = path.join( + path.dirname(fileURLToPath(import.meta.url)), + '..', + '..', + '..', + '..', + '..', + 'docs', + 'vendor', + 'nfx-test-vectors', +); + +interface BindVector { + readonly secret_keys_DO_NOT_USE: Readonly>; + readonly bind: { + readonly seeder: string; + readonly namespace: string; + readonly peer: string; + readonly sig: string; + }; +} + +const v = JSON.parse( + readFileSync(path.join(VECTORS, 'beacon-hyperdht.json'), 'utf8'), +) as BindVector; +const SEEDER = 'a'.repeat(64); +const NS = 'nfx:testnet:1'; + +describe('hyperdht bind', () => { + it('reproduces the vector byte for byte', () => { + const seed = v.secret_keys_DO_NOT_USE['hyperdht_peer']; + if (seed === undefined) throw new Error('vector has no hyperdht_peer seed'); + const keyPair = DHT.keyPair(fromHex(seed)); + expect(toHex(keyPair.publicKey)).toBe(v.bind.peer); + expect(hyperdhtBind(keyPair, v.bind.seeder, v.bind.namespace)).toBe(v.bind.sig); + expect(hyperdhtEndpoint(keyPair, v.bind.seeder, v.bind.namespace)).toEqual({ + t: 'hyperdht', + peer: v.bind.peer, + bind: v.bind.sig, + }); + }); + + it('verifies for its seeder and network only', () => { + const keyPair = DHT.keyPair(new Uint8Array(32).fill(3)); + const e = hyperdhtEndpoint(keyPair, SEEDER, NS); + expect(nfx.verifyHyperdhtBind(e.peer, e.bind, SEEDER, NS)).toBe(true); + expect(nfx.verifyHyperdhtBind(e.peer, e.bind, 'b'.repeat(64), NS)).toBe(false); + expect(nfx.verifyHyperdhtBind(e.peer, e.bind, SEEDER, 'nfx:regtest:1')).toBe(false); + const other = DHT.keyPair(new Uint8Array(32).fill(4)); + expect(nfx.verifyHyperdhtBind(toHex(other.publicKey), e.bind, SEEDER, NS)).toBe(false); + }); + + it('refuses a malformed seeder, namespace or key pair', () => { + const keyPair = DHT.keyPair(new Uint8Array(32).fill(5)); + expect(() => hyperdhtBind(keyPair, 'A'.repeat(64), NS)).toThrow(/invalid-argument/); + expect(() => hyperdhtBind(keyPair, SEEDER, 'nfx:Testnet:1')).toThrow(/invalid-argument/); + expect(() => hyperdhtBind(keyPair, SEEDER, 'not-a-namespace')).toThrow(/invalid-argument/); + const mismatched = { + publicKey: DHT.keyPair(new Uint8Array(32).fill(6)).publicKey, + secretKey: keyPair.secretKey, + }; + expect(() => hyperdhtBind(mismatched, SEEDER, NS)).toThrow(/invalid-argument/); + }); +}); diff --git a/packages/seeder/src/nfx/__tests__/endpoint.integration.test.ts b/packages/seeder/src/nfx/__tests__/endpoint.integration.test.ts new file mode 100644 index 00000000..a57bc0cd --- /dev/null +++ b/packages/seeder/src/nfx/__tests__/endpoint.integration.test.ts @@ -0,0 +1,141 @@ +/** + * NFX-12 over a real HyperDHT: a seeder listens on its peer key, a watcher dials that key and + * fetches a file over `nfx/files/1`, on a LOCAL `hyperdht` testnet (`hyperdht/testnet.js`, + * bootstrap on 127.0.0.1, as the swarm integration tests do). Nothing leaves loopback. + */ +import { nfx } from '@sovit/core'; +import DHT from 'hyperdht'; +import createTestnet from 'hyperdht/testnet.js'; +import type { Testnet } from 'hyperdht/testnet.js'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { nodeCrypto } from '../../adapters/node/crypto.js'; +import { bytesEqual } from '../../util/hex.js'; +import type { NfxConnection } from '../connection.js'; +import { dialNfx, listenNfx, type NfxListener } from '../endpoint.js'; +import { MemorySink, MemorySource, bytes, keyPair, sha256Hex, video } from './helpers.js'; + +vi.setConfig({ testTimeout: 60_000 }); + +const V = video('nfx-dht-video'); +const SEEDER = keyPair(11); +const WATCHER = keyPair(12); +const BANNED = keyPair(13); + +describe('NFX-12 over a local hyperdht testnet', () => { + let testnet: Testnet; + const cleanups: (() => Promise)[] = []; + + beforeEach(async () => { + testnet = await createTestnet(3); + }); + afterEach(async () => { + for (const c of cleanups.splice(0).reverse()) await c(); + await testnet.destroy(); + }); + + function dht(): DHT { + const node = new DHT({ bootstrap: testnet.bootstrap }); + cleanups.push(() => node.destroy()); + return node; + } + + async function seeder( + source: MemorySource, + onConnection?: (c: NfxConnection) => void, + ): Promise { + const node = dht(); + await node.fullyBootstrapped(); + const listener = await listenNfx(node, { + keyPair: SEEDER, + source, + firewall: (remote) => bytesEqual(remote, BANNED.publicKey), + maxConnectionsPerPeer: 1, + ...(onConnection !== undefined ? { onConnection } : {}), + }); + cleanups.push(() => listener.close()); + return listener; + } + + it('dials the seeder by key and fetches a file byte for byte', async () => { + const source = new MemorySource(); + const data = bytes(5 * nfx.MAX_DATA_PAYLOAD + 999, 21); + const sha = source.add(V, data); + const served: NfxConnection[] = []; + await seeder(source, (c) => served.push(c)); + + const watcher = dht(); + await watcher.fullyBootstrapped(); + const dialed = await dialNfx(watcher, SEEDER.publicKey, { + keyPair: WATCHER, + crypto: nodeCrypto, + timeoutMs: 20_000, + }); + if (!dialed.ok) throw new Error(`dial failed: ${dialed.reason}`); + cleanups.push(() => { + dialed.connection.destroy('test over'); + return Promise.resolve(); + }); + expect(bytesEqual(dialed.connection.remotePublicKey, SEEDER.publicKey)).toBe(true); + + const sink = new MemorySink(); + const r = await dialed.connection.files(V)!.fetch({ sha, size: data.byteLength, sink }); + expect(r).toEqual({ ok: true, bytes: data.byteLength }); + expect(sha256Hex(sink.bytes())).toBe(sha); + + // The seeder saw the watcher's dialing key: its account's peer (NFX-12 §1). + expect(served).toHaveLength(1); + expect(bytesEqual(served[0]!.remotePublicKey, WATCHER.publicKey)).toBe(true); + expect(served[0]!.pump.stats.maxQueuedBytes).toBeLessThanOrEqual(nfx.MAX_QUEUED_FILE_BYTES); + }); + + it('refuses a dialer its firewall rejects, and a second connection from one key', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const served: NfxConnection[] = []; + await seeder(source, (c) => served.push(c)); + + const banned = dht(); + await banned.fullyBootstrapped(); + const refused = await dialNfx(banned, SEEDER.publicKey, { + keyPair: BANNED, + crypto: nodeCrypto, + timeoutMs: 15_000, + }); + expect(refused.ok).toBe(false); + + const watcher = dht(); + await watcher.fullyBootstrapped(); + const first = await dialNfx(watcher, SEEDER.publicKey, { + keyPair: WATCHER, + crypto: nodeCrypto, + timeoutMs: 20_000, + }); + if (!first.ok) throw new Error(`dial failed: ${first.reason}`); + cleanups.push(() => { + first.connection.destroy('test over'); + return Promise.resolve(); + }); + await vi.waitFor(() => { + expect(served).toHaveLength(1); + }); + + const again = dht(); + await again.fullyBootstrapped(); + const second = await dialNfx(again, SEEDER.publicKey, { + keyPair: WATCHER, + crypto: nodeCrypto, + timeoutMs: 15_000, + }); + if (second.ok) second.connection.destroy('test over'); + expect(second.ok).toBe(false); + expect(served).toHaveLength(1); + }); + + it('refuses a peer key that is not 32 bytes', async () => { + const watcher = dht(); + expect( + await dialNfx(watcher, new Uint8Array(31), { keyPair: WATCHER, crypto: nodeCrypto }), + ).toEqual({ ok: false, reason: 'invalid-argument' }); + }); +}); diff --git a/packages/seeder/src/nfx/__tests__/helpers.ts b/packages/seeder/src/nfx/__tests__/helpers.ts new file mode 100644 index 00000000..b5cbc3c0 --- /dev/null +++ b/packages/seeder/src/nfx/__tests__/helpers.ts @@ -0,0 +1,148 @@ +/** + * Test helpers for the NFX-12 transport: an in-memory file source and sink, and a pair of + * real secret-streams joined in memory (`@hyperswarm/secret-stream` without a transport), so + * most tests run the real Noise + Protomux stack without a DHT. + */ +import { createHash } from 'node:crypto'; + +import NoiseSecretStream from '@hyperswarm/secret-stream'; +import { nfx } from '@sovit/core'; +import DHT from 'hyperdht'; + +import { nodeCrypto } from '../../adapters/node/crypto.js'; +import { NfxConnection, type NfxLimits, type NfxServeOptions } from '../connection.js'; +import type { NfxSink } from '../files-client.js'; +import type { NfxFileSource, NfxServedFile } from '../files-server.js'; + +export const sha256Hex = (data: Uint8Array): string => + createHash('sha256').update(data).digest('hex'); + +export function video(id: string, namespace = 'nfx:regtest:1'): nfx.VideoAddr { + const v = nfx.VideoAddr.parse(`${namespace}:${id}`); + if (!v.ok) throw new Error(`bad test video ${id}`); + return v.value; +} + +/** `n` deterministic bytes. */ +export function bytes(n: number, seed = 1): Uint8Array { + const out = new Uint8Array(n); + for (let i = 0; i < n; i++) out[i] = (i * 31 + seed * 7) & 0xff; + return out; +} + +export function keyPair(seed: number): { publicKey: Uint8Array; secretKey: Uint8Array } { + return DHT.keyPair(new Uint8Array(32).fill(seed)); +} + +/** Files per video, by sha. `add` may file bytes under a sha that is not theirs (poison). */ +export class MemorySource implements NfxFileSource { + private readonly byVideo = new Map>(); + /** Every video counts as served (for the video-limit test). */ + allVideos = false; + /** While set, `file()` waits for it (holds requests open). */ + gate: Promise | null = null; + /** Reads at or past this offset never resolve (a transfer that stops mid-file). */ + hangReadsFrom: number | null = null; + readonly lookups: string[] = []; + + add(v: nfx.VideoAddr, data: Uint8Array, sha = sha256Hex(data)): string { + const key = v.toString(); + const files = this.byVideo.get(key) ?? new Map(); + files.set(sha, data); + this.byVideo.set(key, files); + return sha; + } + + hasVideo(v: nfx.VideoAddr): boolean { + return this.allVideos || this.byVideo.has(v.toString()); + } + + async file(v: nfx.VideoAddr, sha: string): Promise { + this.lookups.push(`${v.toString()} ${sha}`); + if (this.gate !== null) await this.gate; + const data = this.byVideo.get(v.toString())?.get(sha); + if (data === undefined) return undefined; + const hang = this.hangReadsFrom; + return { + size: data.byteLength, + read: (offset, length) => + hang !== null && offset >= hang + ? new Promise(() => undefined) + : Promise.resolve(data.slice(offset, offset + length)), + }; + } +} + +export class MemorySink implements NfxSink { + readonly chunks: Uint8Array[] = []; + + write(chunk: Uint8Array): void { + this.chunks.push(chunk); + } + + bytes(): Uint8Array { + const out = new Uint8Array(this.chunks.reduce((n, c) => n + c.byteLength, 0)); + let at = 0; + for (const c of this.chunks) { + out.set(c, at); + at += c.byteLength; + } + return out; + } +} + +/** Two opened secret-streams joined in memory: `client` dialed `server`. */ +export async function streamPair( + serverSeed = 1, + clientSeed = 2, +): Promise<{ server: NoiseSecretStream; client: NoiseSecretStream }> { + const client = new NoiseSecretStream(true, null, { keyPair: keyPair(clientSeed) }); + const server = new NoiseSecretStream(false, null, { keyPair: keyPair(serverSeed) }); + client.rawStream.pipe(server.rawStream).pipe(client.rawStream); + client.on('error', () => undefined); + server.on('error', () => undefined); + const [a, b] = await Promise.all([client.opened, server.opened]); + if (!a || !b) throw new Error('the in-memory pair did not open'); + return { server, client }; +} + +export interface ConnectedPair { + readonly server: NfxConnection; + readonly client: NfxConnection; + readonly serverClosed: Promise; +} + +/** A seeder connection serving `serve` and a watcher connection, joined in memory. */ +export async function connectedPair( + serve: NfxServeOptions, + limits?: Partial, +): Promise { + const s = await streamPair(); + let closed: (reason: string) => void = () => undefined; + const serverClosed = new Promise((resolve) => { + closed = resolve; + }); + const server = new NfxConnection(s.server, { + serve, + ...(limits !== undefined ? { limits } : {}), + onClose: (reason) => { + closed(reason); + }, + }); + const client = new NfxConnection(s.client, { + crypto: nodeCrypto, + ...(limits !== undefined ? { limits } : {}), + }); + return { server, client, serverClosed }; +} + +/** Resolves when `socket` closes. */ +export function closedOf(socket: NoiseSecretStream): Promise { + return new Promise((resolve) => { + if (socket.destroyed) { + resolve(); + return; + } + socket.once('close', resolve); + }); +} diff --git a/packages/seeder/src/nfx/__tests__/pump.test.ts b/packages/seeder/src/nfx/__tests__/pump.test.ts new file mode 100644 index 00000000..357f6163 --- /dev/null +++ b/packages/seeder/src/nfx/__tests__/pump.test.ts @@ -0,0 +1,106 @@ +/** + * The per-connection pump (NFX-12 §4 flow control) against a stand-in for a streamx stream: + * `write` returns `false` once 16 KiB are buffered, and `'drain'` fires when a buffer that + * asked for one has emptied. The real queue never exceeds one frame on a sub-HWM buffer. + */ +import { EventEmitter } from 'node:events'; + +import { nfx } from '@sovit/core'; +import { describe, expect, it } from 'vitest'; + +import { FilePump } from '../pump.js'; + +const HWM = 16_384; +const FRAME = nfx.MAX_DATA_PAYLOAD + 4; + +class FakeStream extends EventEmitter { + buffered = 0; + maxBuffered = 0; + private needDrain = false; + + write(n: number): boolean { + this.buffered += n; + this.maxBuffered = Math.max(this.maxBuffered, this.buffered); + const below = this.buffered < HWM; + if (!below) this.needDrain = true; + return below; + } + + /** The transport takes `n` bytes. */ + flush(n: number): void { + this.buffered = Math.max(0, this.buffered - n); + if (this.buffered === 0 && this.needDrain) { + this.needDrain = false; + this.emit('drain'); + } + } +} + +const tick = (): Promise => new Promise((r) => setImmediate(r)); + +describe('FilePump', () => { + it('writes one frame on a full buffer at most, and resumes on drain', async () => { + const stream = new FakeStream(); + const pump = new FilePump(stream); + const order: string[] = []; + let busy = 0; + const transfer = async (name: string, frames: number): Promise => { + for (let i = 0; i < frames; i++) { + if (!(await pump.turn())) return; + busy++; + expect(busy).toBe(1); // one turn out at a time + expect(stream.buffered).toBeLessThan(HWM); // never written on a buffer asking for a drain + order.push(name); + pump.sent(FRAME, stream.write(FRAME)); + busy--; + } + }; + const all = Promise.all([transfer('a', 6), transfer('b', 6), transfer('c', 6)]); + for (let i = 0; i < 400 && order.length < 18; i++) { + await tick(); + stream.flush(8_192); + } + await all; + expect(order).toHaveLength(18); + // Arrival order: the three transfers interleave instead of one draining first. + expect(order.slice(0, 6)).toEqual(['a', 'b', 'c', 'a', 'b', 'c']); + expect(stream.maxBuffered).toBeLessThanOrEqual(HWM - 1 + FRAME); + expect(pump.stats.frames).toBe(18); + expect(pump.stats.stalls).toBe(18); + expect(pump.stats.maxQueuedBytes).toBeLessThanOrEqual(nfx.MAX_QUEUED_FILE_BYTES); + expect(pump.stats.maxQueuedBytes).toBeGreaterThanOrEqual(stream.maxBuffered); + }); + + it('keeps writing small frames while the stream accepts them', async () => { + const stream = new FakeStream(); + const pump = new FilePump(stream); + for (let i = 0; i < 10; i++) { + expect(await pump.turn()).toBe(true); + pump.sent(100, stream.write(100)); + stream.flush(100); + } + expect(pump.stats.stalls).toBe(0); + expect(pump.stats.maxQueuedBytes).toBeLessThan(HWM); + }); + + it('gives a skipped turn to the next transfer', async () => { + const pump = new FilePump(new FakeStream()); + expect(await pump.turn()).toBe(true); + const second = pump.turn(); + pump.skip(); + expect(await second).toBe(true); + }); + + it('answers every waiter `false` once the stream closes', async () => { + const stream = new FakeStream(); + const pump = new FilePump(stream); + expect(await pump.turn()).toBe(true); + pump.sent(FRAME, stream.write(FRAME)); // blocked now + const waiting = [pump.turn(), pump.turn()]; + stream.emit('close'); + expect(await Promise.all(waiting)).toEqual([false, false]); + expect(await pump.turn()).toBe(false); + expect(stream.listenerCount('drain')).toBe(0); + expect(stream.listenerCount('close')).toBe(0); + }); +}); diff --git a/packages/seeder/src/nfx/__tests__/transport.test.ts b/packages/seeder/src/nfx/__tests__/transport.test.ts new file mode 100644 index 00000000..1ad208ee --- /dev/null +++ b/packages/seeder/src/nfx/__tests__/transport.test.ts @@ -0,0 +1,460 @@ +/** + * NFX-12 `nfx/files/1` and `nfx/pay/1` end to end over a real secret-stream pair joined in + * memory: the seeder side (`files-server.ts`), the watcher side (`files-client.ts`), and the + * connection's rules (`connection.ts`). Hostile peers speak raw Protomux. + */ +import { nfx } from '@sovit/core'; +import Protomux from 'protomux'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { nodeCrypto } from '../../adapters/node/crypto.js'; +import { fromHex } from '../../util/hex.js'; +import { NfxConnection } from '../connection.js'; +import { TRAPPED_TYPES } from '../traps.js'; +import { + MemorySink, + MemorySource, + bytes, + closedOf, + connectedPair, + sha256Hex, + streamPair, + video, + type ConnectedPair, +} from './helpers.js'; + +const V = video('nfx-transport-a'); +const W = video('nfx-transport-b'); + +const open: ConnectedPair[] = []; +afterEach(() => { + for (const p of open.splice(0)) { + p.client.destroy('test over'); + p.server.destroy('test over'); + } +}); + +async function pair(source: MemorySource, limits?: Parameters[1]) { + const p = await connectedPair({ source }, limits); + open.push(p); + return p; +} + +/** A `get` frame as a hostile client would send it. */ +function get(req: number, sha: string): Uint8Array { + const f = nfx.encodeFilesMessage({ type: 'get', req, sha: fromHex(sha) }); + if (!f.ok) throw new Error(f.error.reason); + return f.value.bytes; +} + +describe('nfx/files/1 over a secret-stream pair', () => { + it('fetches a multi-frame file byte for byte, its sha256 checked', async () => { + const source = new MemorySource(); + const data = bytes(3 * nfx.MAX_DATA_PAYLOAD + 123); + const sha = source.add(V, data); + const p = await pair(source); + const sink = new MemorySink(); + const r = await p.client.files(V)!.fetch({ sha, size: data.byteLength, sink }); + expect(r).toEqual({ ok: true, bytes: data.byteLength }); + expect(sink.chunks).toHaveLength(4); + expect(sha256Hex(sink.bytes())).toBe(sha); + expect(p.server.pump.stats.frames).toBe(4); + }); + + it('fetches the hash list against a cap instead of an exact size', async () => { + const source = new MemorySource(); + const list = bytes(5000, 9); + const root = source.add(V, list); + const p = await pair(source); + const sink = new MemorySink(); + const r = await p.client.files(V)!.fetch({ + sha: root, + size: nfx.MAX_HASH_LIST_BYTES, + exact: false, + sink, + }); + expect(r).toEqual({ ok: true, bytes: 5000 }); + }); + + it('serves a zero-byte file as `end` alone', async () => { + const source = new MemorySource(); + const sha = source.add(V, new Uint8Array(0)); + const p = await pair(source); + const sink = new MemorySink(); + expect(await p.client.files(V)!.fetch({ sha, size: 0, sink })).toEqual({ ok: true, bytes: 0 }); + expect(sink.chunks).toHaveLength(0); + }); + + it('answers `missing` for a file it does not list for THIS video', async () => { + const source = new MemorySource(); + const onlyOnW = source.add(W, bytes(100, 3)); + source.add(V, bytes(10, 4)); + const p = await pair(source); + const sink = new MemorySink(); + const r = await p.client.files(V)!.fetch({ sha: onlyOnW, size: 100, sink }); + // Held for W, asked on V's channel: never an oracle for the rest of the store (§4). + expect(r).toEqual({ ok: false, reason: 'missing' }); + expect(source.lookups).toEqual([`${V.toString()} ${onlyOnW}`]); + expect(p.client.closed).toBe(false); + }); + + it('reports a video the seeder does not serve as `unserved`', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const p = await pair(source); + const r = await p.client + .files(W)! + .fetch({ sha: 'ab'.repeat(32), size: 1, sink: new MemorySink() }); + expect(r).toEqual({ ok: false, reason: 'unserved' }); + }); + + it('rejects a poisoned file and drops the source', async () => { + const source = new MemorySource(); + const real = bytes(70_000, 5); + const sha = sha256Hex(real); + source.add(V, bytes(70_000, 6), sha); // same size, other bytes + const p = await pair(source); + const r = await p.client + .files(V)! + .fetch({ sha, size: real.byteLength, sink: new MemorySink() }); + expect(r).toEqual({ ok: false, reason: 'mismatch' }); + await closedOf(p.client.socket); + expect(p.client.closed).toBe(true); + }); + + it('refuses more bytes than the size, and `end` before it', async () => { + const source = new MemorySource(); + const data = bytes(1000, 7); + const sha = source.add(V, data); + const long = await pair(source); + expect(await long.client.files(V)!.fetch({ sha, size: 999, sink: new MemorySink() })).toEqual({ + ok: false, + reason: 'too-long', + }); + await closedOf(long.client.socket); + const short = await pair(source); + expect(await short.client.files(V)!.fetch({ sha, size: 1001, sink: new MemorySink() })).toEqual( + { ok: false, reason: 'too-short' }, + ); + await closedOf(short.client.socket); + }); + + it('refuses bad arguments, a second request for the same sha, and a ninth request', async () => { + const source = new MemorySource(); + const a = source.add(V, bytes(10, 1)); + let release: () => void = () => undefined; + source.gate = new Promise((r) => { + release = r; + }); + const p = await pair(source); + const files = p.client.files(V)!; + const sink = new MemorySink(); + expect(await files.fetch({ sha: 'AB'.repeat(32), size: 1, sink })).toEqual({ + ok: false, + reason: 'invalid-argument', + }); + const first = files.fetch({ sha: a, size: 10, sink }); + expect(await files.fetch({ sha: a, size: 10, sink })).toEqual({ + ok: false, + reason: 'duplicate', + }); + const more = Array.from({ length: 7 }, (_, i) => + files.fetch({ sha: (i + 10).toString(16).padStart(2, '0').repeat(32), size: 1, sink }), + ); + await vi.waitFor(() => { + expect(files.openRequests).toBe(8); + }); + expect(await files.fetch({ sha: 'ee'.repeat(32), size: 1, sink })).toEqual({ + ok: false, + reason: 'busy', + }); + release(); + expect(await first).toEqual({ ok: true, bytes: 10 }); + for (const r of await Promise.all(more)) expect(r).toEqual({ ok: false, reason: 'missing' }); + }); + + it('spools to a slow sink with reads held, the seeder queueing at most 256 KiB', async () => { + const source = new MemorySource(); + const data = bytes(2 * 1024 * 1024 + 17, 8); + const sha = source.add(V, data); + const p = await pair(source); + const hold = vi.spyOn(p.client, 'holdReads'); + const chunks: Uint8Array[] = []; + const slow = { + async write(chunk: Uint8Array): Promise { + chunks.push(chunk); + await new Promise((r) => setTimeout(r, 5)); // far below an in-memory pair's rate + }, + }; + const r = await p.client.files(V)!.fetch({ sha, size: data.byteLength, sink: slow }); + expect(r).toEqual({ ok: true, bytes: data.byteLength }); + expect(hold).toHaveBeenCalled(); + expect(p.server.pump.stats.maxQueuedBytes).toBeLessThanOrEqual(nfx.MAX_QUEUED_FILE_BYTES); + expect(p.server.pump.stats.stalls).toBeGreaterThan(0); + }, 30_000); +}); + +describe('hostile watchers', () => { + /** A seeder connection and a raw Protomux on the watcher's stream. */ + async function hostile(source: MemorySource) { + const s = await streamPair(); + let reason = ''; + const server = new NfxConnection(s.server, { + serve: { source }, + onClose: (why) => { + reason = why; + }, + }); + const mux = Protomux.from(s.client); + const channel = (v: nfx.VideoAddr, onclose?: () => void) => { + const ch = mux.createChannel({ + protocol: nfx.FILES_PROTOCOL, + id: v.channelId(), + onclose: () => { + onclose?.(); + }, + })!; + const messages = Array.from({ length: 4 + TRAPPED_TYPES + 1 }, () => ch.addMessage({})); + ch.open(); + return { ch, messages }; + }; + return { server, client: s.client, channel, reason: () => reason }; + } + + it('closes the channel on an unknown message type', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const h = await hostile(source); + let closed = false; + const { ch, messages } = h.channel(V, () => { + closed = true; + }); + expect(await ch.fullyOpened()).toBe(true); + messages[4]!.send(new Uint8Array(4)); // index 4: the first trap + await vi.waitFor(() => { + expect(closed).toBe(true); + }); + h.server.destroy('test over'); + }); + + it('closes the channel on a `req` that is already open', async () => { + const source = new MemorySource(); + const sha = source.add(V, bytes(10)); + source.gate = new Promise(() => undefined); // never answers + const h = await hostile(source); + let closed = false; + const { ch, messages } = h.channel(V, () => { + closed = true; + }); + expect(await ch.fullyOpened()).toBe(true); + messages[0]!.send(get(7, sha)); + messages[0]!.send(get(7, sha)); + await vi.waitFor(() => { + expect(closed).toBe(true); + }); + h.server.destroy('test over'); + }); + + it('closes the connection on a ninth open request on one channel', async () => { + const source = new MemorySource(); + source.gate = new Promise(() => undefined); + source.add(V, bytes(10)); + const h = await hostile(source); + const { ch, messages } = h.channel(V); + expect(await ch.fullyOpened()).toBe(true); + for (let req = 1; req <= 9; req++) messages[0]!.send(get(req, 'cd'.repeat(32))); + await closedOf(h.client); + expect(h.reason()).toMatch(/too many open requests/); + }); + + it('closes the connection past 16 videos', async () => { + const source = new MemorySource(); + source.allVideos = true; + const h = await hostile(source); + for (let i = 0; i < 17; i++) h.channel(video(`nfx-video-${String(i).padStart(3, '0')}`)); + await closedOf(h.client); + expect(h.reason()).toMatch(/more videos/); + }); + + it('closes the connection after 64 rejected channel opens', async () => { + const h = await hostile(new MemorySource()); // serves nothing + const mux = Protomux.from(h.client); + for (let i = 0; i <= 64; i++) { + // Not canonical (uppercase), so no video is claimed: only the rejection budget counts. + mux + .createChannel({ + protocol: nfx.FILES_PROTOCOL, + id: new TextEncoder().encode(`NFX:regtest:1:bad-${String(i)}`), + }) + ?.open(); + } + await closedOf(h.client); + expect(h.reason()).toMatch(/rejected channel opens/); + }); + + it('rejects a channel id that is not a canonical video address', async () => { + const source = new MemorySource(); + source.allVideos = true; + const s = await streamPair(); + const server = new NfxConnection(s.server, { serve: { source } }); + const mux = Protomux.from(s.client); + const ch = mux.createChannel({ + protocol: nfx.FILES_PROTOCOL, + id: new TextEncoder().encode('NFX:regtest:1:Not-Canonical'), + })!; + ch.open(); + expect(await ch.fullyOpened()).toBe(false); + server.destroy('test over'); + }); +}); + +describe('the watcher side', () => { + it('opens no more than 16 videos on one connection', async () => { + const p = await pair(new MemorySource()); + for (let i = 0; i < 16; i++) { + expect(p.client.files(video(`nfx-watch-${String(i).padStart(3, '0')}`))).toBeDefined(); + } + expect(p.client.files(video('nfx-watch-one-more'))).toBeUndefined(); + // The same video again is the same client, not a new one. + expect(p.client.files(video('nfx-watch-000'))).toBe(p.client.files(video('nfx-watch-000'))); + }); + + it('times out a fetch and abandons it by closing the channel', async () => { + const source = new MemorySource(); + const sha = source.add(V, bytes(10)); + source.gate = new Promise(() => undefined); + const p = await pair(source); + const files = p.client.files(V)!; + const r = await files.fetch({ sha, size: 10, sink: new MemorySink(), timeoutMs: 50 }); + expect(r).toEqual({ ok: false, reason: 'timeout' }); + expect(files.closed).toBe(true); + }); + + it('abandons a transfer that stops while keepalives hold the connection open', async () => { + const source = new MemorySource(); + const data = bytes(3 * nfx.MAX_DATA_PAYLOAD, 12); + const sha = source.add(V, data); + source.hangReadsFrom = nfx.MAX_DATA_PAYLOAD; // the second frame never comes + const p = await pair(source, { keepAliveMs: 40, idleTimeoutMs: 300 }); + const files = p.client.files(V)!; + const r = await files.fetch({ sha, size: data.byteLength, sink: new MemorySink() }); + expect(r).toEqual({ ok: false, reason: 'timeout' }); + expect(files.closed).toBe(true); + expect(p.client.closed).toBe(false); + }); + + it('fails a fetch whose sink throws', async () => { + const source = new MemorySource(); + const sha = source.add(V, bytes(10)); + const p = await pair(source); + const sink = { + write(): void { + throw new Error('disk full'); + }, + }; + expect(await p.client.files(V)!.fetch({ sha, size: 10, sink })).toEqual({ + ok: false, + reason: 'sink', + }); + }); +}); + +describe('liveness (short intervals in place of 25 s and 60 s)', () => { + it('keeps an idle connection open with keepalives', async () => { + const p = await pair(new MemorySource(), { keepAliveMs: 40, idleTimeoutMs: 200 }); + await new Promise((r) => setTimeout(r, 700)); + expect(p.server.closed).toBe(false); + expect(p.client.closed).toBe(false); + }); + + it('closes a connection that has been silent for the idle timeout', async () => { + const s = await streamPair(); + let reason = ''; + const server = new NfxConnection(s.server, { + serve: { source: new MemorySource() }, + limits: { keepAliveMs: 40, idleTimeoutMs: 200 }, + onClose: (why) => { + reason = why; + }, + }); + // The watcher never writes and sends no keepalives. + await closedOf(s.server); + expect(server.closed).toBe(true); + expect(reason).not.toBe(''); + s.client.destroy(); + }); +}); + +describe('nfx/pay/1 lane', () => { + it('carries messages both ways within the codec rules', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const p = await connectedPair({ + source, + onPay: () => ({ + onMessage: (text, lane) => { + lane.send(`echo:${text}`); + }, + }), + }); + open.push(p); + const got: string[] = []; + const lane = p.client.pay(V, { + onMessage: (text) => { + got.push(text); + }, + })!; + expect(await lane.opened()).toBe(true); + expect(lane.send('{"t":"hello"}')).toEqual({ ok: true, value: true }); + await vi.waitFor(() => { + expect(got).toEqual(['echo:{"t":"hello"}']); + }); + expect(lane.send('two\nlines').ok).toBe(false); + expect(lane.send('x'.repeat(nfx.MAX_PAY_MESSAGE + 1)).ok).toBe(false); + }); + + it('is rejected by a seeder in free mode', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const p = await pair(source); + const lane = p.client.pay(V, { onMessage: () => undefined })!; + expect(await lane.opened()).toBe(false); + }); + + it('closes on a message that breaks the rules', async () => { + const source = new MemorySource(); + source.add(V, bytes(10)); + const errors: string[] = []; + const s = await streamPair(); + const server = new NfxConnection(s.server, { + serve: { + source, + onPay: () => ({ + onMessage: () => undefined, + onProtocolError: (why) => { + errors.push(why); + }, + }), + }, + }); + const mux = Protomux.from(s.client); + const ch = mux.createChannel({ protocol: nfx.PAY_PROTOCOL, id: V.channelId() })!; + const msg = ch.addMessage({}); + ch.open(); + expect(await ch.fullyOpened()).toBe(true); + msg.send(new TextEncoder().encode('a\nb')); + await vi.waitFor(() => { + expect(errors).toHaveLength(1); + }); + server.destroy('test over'); + }); +}); + +describe('connection', () => { + it('refuses a stream whose handshake has not completed', () => { + const fake = { remotePublicKey: null } as unknown as ConstructorParameters< + typeof NfxConnection + >[0]; + expect(() => new NfxConnection(fake, { crypto: nodeCrypto })).toThrow(/not open/); + }); +}); From 4536a718f3d96432e2985e80faae270a6a31b284 Mon Sep 17 00:00:00 2001 From: sovtech Date: Thu, 22 Oct 2026 17:40:42 -0600 Subject: [PATCH 4/5] =?UTF-8?q?fix(seeder/nfx):=20review=20=E2=80=94=20a?= =?UTF-8?q?=20BOM-prefixed=20channel=20id,=20and=20data=20after=20`end`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - connection.ts: the channel-id check decoded with TextDecoder's default, which strips a leading U+FEFF, so `` + a served address passed the canonical-form comparison although its bytes are not the canonical id (the channel was then opened under the canonical id instead). `ignoreBOM: true` keeps it, and the exact comparison refuses it. - files-client.ts: `data` or a second `end` for a request whose `end` already arrived was taken as more of the file (hashed and written to the sink) while the sha256 check waited for the sink. It is now a protocol error, and the deferred check does nothing once the request is settled (so it can no longer drop a connection for a transfer already failed). - Tests: a BOM-prefixed id is refused; `data` corked behind `end` in one write fails the hash-list fetch `protocol`. holepunch.d.ts declares Protomux `cork`/`uncork` for that test. --- .../src/nfx/__tests__/transport.test.ts | 49 ++++++++++++++++++- packages/seeder/src/nfx/connection.ts | 3 +- packages/seeder/src/nfx/files-client.ts | 17 +++++-- packages/seeder/src/types/holepunch.d.ts | 3 ++ 4 files changed, 65 insertions(+), 7 deletions(-) diff --git a/packages/seeder/src/nfx/__tests__/transport.test.ts b/packages/seeder/src/nfx/__tests__/transport.test.ts index 1ad208ee..cdc46297 100644 --- a/packages/seeder/src/nfx/__tests__/transport.test.ts +++ b/packages/seeder/src/nfx/__tests__/transport.test.ts @@ -4,7 +4,7 @@ * connection's rules (`connection.ts`). Hostile peers speak raw Protomux. */ import { nfx } from '@sovit/core'; -import Protomux from 'protomux'; +import Protomux, { type ProtomuxMessage } from 'protomux'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { nodeCrypto } from '../../adapters/node/crypto.js'; @@ -304,6 +304,13 @@ describe('hostile watchers', () => { })!; ch.open(); expect(await ch.fullyOpened()).toBe(false); + // A served address behind a byte-order mark: the bytes are not the canonical id. + const bom = mux.createChannel({ + protocol: nfx.FILES_PROTOCOL, + id: new Uint8Array([0xef, 0xbb, 0xbf, ...V.channelId()]), + })!; + bom.open(); + expect(await bom.fullyOpened()).toBe(false); server.destroy('test over'); }); }); @@ -343,6 +350,46 @@ describe('the watcher side', () => { expect(p.client.closed).toBe(false); }); + it('treats `data` after `end` as a protocol error, not part of the file', async () => { + const data = bytes(10, 13); + const sha = sha256Hex(data); + const s = await streamPair(); + const client = new NfxConnection(s.client, { crypto: nodeCrypto }); + const mux = Protomux.from(s.server); + const frame = (m: nfx.FilesMessage): Uint8Array => { + const f = nfx.encodeFilesMessage(m); + if (!f.ok) throw new Error(f.error.reason); + return f.value.bytes; + }; + mux.pair({ protocol: nfx.FILES_PROTOCOL }, () => { + const ch = mux.createChannel({ protocol: nfx.FILES_PROTOCOL, id: V.channelId() })!; + const out: { data?: ProtomuxMessage; end?: ProtomuxMessage } = {}; + ch.addMessage({ + onmessage: (b) => { + const g = nfx.decodeFilesMessage(0, b); + if (!g.ok) return; + const { req } = g.value; + // One write, so the three frames arrive together, before the sha256 check runs. + mux.cork(); + out.data!.send(frame({ type: 'data', req, payload: data })); + out.end!.send(frame({ type: 'end', req })); + out.data!.send(frame({ type: 'data', req, payload: bytes(5, 14) })); + mux.uncork(); + }, + }); + out.data = ch.addMessage({}); + out.end = ch.addMessage({}); + ch.addMessage({}); + ch.open(); + }); + // A hash-list fetch (a cap, not an exact size), so the late bytes are not `too-long`. + const r = await client + .files(V)! + .fetch({ sha, size: 100, exact: false, sink: new MemorySink() }); + expect(r).toEqual({ ok: false, reason: 'protocol' }); + client.destroy('test over'); + }); + it('fails a fetch whose sink throws', async () => { const source = new MemorySource(); const sha = source.add(V, bytes(10)); diff --git a/packages/seeder/src/nfx/connection.ts b/packages/seeder/src/nfx/connection.ts index 916b0533..ec1ca09c 100644 --- a/packages/seeder/src/nfx/connection.ts +++ b/packages/seeder/src/nfx/connection.ts @@ -217,7 +217,8 @@ export class NfxConnection implements ConnectionHost { if (id === null || id.byteLength > 256 || this.closed) return undefined; let text: string; try { - text = new TextDecoder('utf-8', { fatal: true }).decode(id); + // `ignoreBOM` keeps a leading U+FEFF, so the canonical-form check below refuses it. + text = new TextDecoder('utf-8', { fatal: true, ignoreBOM: true }).decode(id); } catch { return undefined; } diff --git a/packages/seeder/src/nfx/files-client.ts b/packages/seeder/src/nfx/files-client.ts index 4c4a714c..90bd25b2 100644 --- a/packages/seeder/src/nfx/files-client.ts +++ b/packages/seeder/src/nfx/files-client.ts @@ -9,8 +9,9 @@ * sha256 is checked before the result says `ok`, and a mismatch drops the whole connection: * the source is discarded for the session (NFX-05 §4). The sink's bytes are the caller's to * discard on any failure; only `ok` means they are the file. - * - `data`, `end` or `missing` for a `req` that is not open, a seeder's `get`, or a trapped - * unknown type close the channel; its open requests fail `protocol`. + * - `data`, `end` or `missing` for a `req` that is not open, `data` or a second `end` after a + * request's `end`, a seeder's `get`, or a trapped unknown type close the channel; its open + * requests fail `protocol`. * - A request that gets no frame for the idle limit (60 s) fails `timeout`, and the channel is * closed to abandon it: a seeder's keepalives hold the connection open, so the connection's * own idle rule cannot catch a transfer that simply stops. @@ -81,6 +82,8 @@ interface Transfer { readonly hasher: Sha256Hasher; received: number; waiting: number; + /** `end` arrived; the sha256 check waits for the sink. */ + ended: boolean; /** Restart the stall timer: a frame of this request arrived. */ alive(): void; writes: Promise; @@ -197,6 +200,7 @@ export class FilesClient { hasher: this.crypto.createSha256(), received: 0, waiting: 0, + ended: false, alive: () => { if (stall !== undefined) clearTimeout(stall); stall = setTimeout(() => { @@ -232,7 +236,7 @@ export class FilesClient { const m = this.decode(1, bytes); if (m?.type !== 'data') return; const t = this.open.get(m.req); - if (t === undefined) { + if (t === undefined || t.ended) { this.protocolError(); return; } @@ -265,16 +269,19 @@ export class FilesClient { const m = this.decode(2, bytes); if (m?.type !== 'end') return; const t = this.open.get(m.req); - if (t === undefined) { + if (t === undefined || t.ended) { this.protocolError(); return; } + t.ended = true; if (t.exact && t.received !== t.size) { this.discardSource(t, 'too-short'); return; } + const { req } = m; void t.writes.then(() => { - if (t.sinkFailed) return; + // Settled meanwhile (a protocol error, a timeout): nothing left to check. + if (t.sinkFailed || this.open.get(req) !== t) return; if (t.hasher.digestHex() !== t.sha) { this.discardSource(t, 'mismatch'); return; diff --git a/packages/seeder/src/types/holepunch.d.ts b/packages/seeder/src/types/holepunch.d.ts index 7c1e50e7..7b43a74d 100644 --- a/packages/seeder/src/types/holepunch.d.ts +++ b/packages/seeder/src/types/holepunch.d.ts @@ -367,6 +367,9 @@ declare module 'protomux' { opts: { readonly protocol: string; readonly id?: Uint8Array | null }, notify: (id: Uint8Array | null) => unknown, ): void; + /** Batch the messages sent until `uncork` into one stream write (tests: one arrival). */ + cork(): void; + uncork(): void; } export default Protomux; } From ab8017667c273fe8c68ba3c5eba5248bf63b9ccc Mon Sep 17 00:00:00 2001 From: sovtech Date: Thu, 22 Oct 2026 17:45:13 -0600 Subject: [PATCH 5/5] =?UTF-8?q?fix(seeder/nfx):=20the=20first=20CI=20findi?= =?UTF-8?q?ngs=20=E2=80=94=20a=20narrowed=20channel=20in=20a=20hoisted=20f?= =?UTF-8?q?unction,=20and=20closed=20read=20after=20an=20await?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tsc: `ch` was narrowed non-null after its check, but not inside the hoisted `serve`, so it is bound once as a ProtomuxChannel. eslint no-unnecessary-condition: `ch.closed` and `this.closed` were re-checked after an await, which TypeScript still narrows to false from the earlier check; both are now read through a function, so the re-check stays (the channel or connection can close while the read is pending). --- packages/seeder/src/nfx/connection.ts | 7 ++++++- packages/seeder/src/nfx/files-server.ts | 12 ++++++++---- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/packages/seeder/src/nfx/connection.ts b/packages/seeder/src/nfx/connection.ts index ec1ca09c..59a0d0de 100644 --- a/packages/seeder/src/nfx/connection.ts +++ b/packages/seeder/src/nfx/connection.ts @@ -234,6 +234,11 @@ export class NfxConnection implements ConnectionHost { } catch { served = false; } - return served && !this.closed ? video.value : undefined; + return served && this.isOpen() ? video.value : undefined; + } + + /** `!closed`, read fresh after an `await` (the earlier check's narrowing no longer holds). */ + private isOpen(): boolean { + return !this.closed; } } diff --git a/packages/seeder/src/nfx/files-server.ts b/packages/seeder/src/nfx/files-server.ts index 59fe8cda..f31cdedc 100644 --- a/packages/seeder/src/nfx/files-server.ts +++ b/packages/seeder/src/nfx/files-server.ts @@ -72,8 +72,12 @@ export function serveFiles( video: nfx.VideoAddr, o: FilesServerOptions, ): ProtomuxChannel | null { - const ch = host.mux.createChannel({ protocol: nfx.FILES_PROTOCOL, id: video.channelId() }); - if (ch === null) return null; + const created = host.mux.createChannel({ protocol: nfx.FILES_PROTOCOL, id: video.channelId() }); + if (created === null) return null; + // Bound once: TypeScript does not carry the null check into the hoisted `serve` below. + const ch: ProtomuxChannel = created; + // Read fresh after each `await`: the channel can close while a read or the source is pending. + const isClosed = (): boolean => ch.closed; const open = new Set(); const fail = (why: string): void => { o.onProtocolError?.(why); @@ -138,7 +142,7 @@ export function serveFiles( fail('the source failed'); return; } - if (ch.closed) return; + if (isClosed()) return; if (file === undefined) { send(missing, { type: 'missing', req }); return; @@ -159,7 +163,7 @@ export function serveFiles( fail('the source failed mid-file'); return; } - if (ch.closed) { + if (isClosed()) { host.pump.skip(); return; }