feat(seeder/nfx): the NFX-12 HyperDHT transport — dial, listen, files/1 both sides, pay/1 lane

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
 

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.

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"
   }

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<Record<string, string>>;
+  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/);
+  });
+});

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<void>)[] = [];
+
+  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<NfxListener> {
+    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' });
+  });
+});