hyper-proto-peer / test / serve-device.test.js

// The control surface: a client/app submits a write to a running SwarmDevice and observes it, over
// the real swarm RPC (serveDevice), gated by the canWrite allowlist — the multi-device analog of
// lib/peer.js apply-write. Isolated from the (separately-proven, multi-device.test.js) attestation
// admission path by backing the device with an in-memory KV, so this test targets ONLY the
// transport: get-state (always), apply-write (authorized lands; unauthorized rejected).

import '../lib/bare-globals.js'
import test from 'brittle'
import crypto from 'hypercore-crypto'
import b4a from 'b4a'
import c from 'compact-encoding'
import Hyperswarm from 'hyperswarm'
import ProtomuxRPC from 'protomux-rpc'
import createTestnet from '@hyperswarm/testnet'
import { SwarmDevice } from '../lib/swarm-device.js'
import { serveDevice, DEVICE_PROTOCOL } from '../lib/serve-device.js'

const DID = 'did:plc:control22222222222222222'
const POST = 'app.bsky.feed.post'

// in-memory KV statebase — RepoBase/Lease only need async set/del/get/list.
function memKV () {
  const m = new Map()
  return {
    async set (k, v) { m.set(k, v) },
    async del (k) { m.delete(k) },
    async get (k) { return m.has(k) ? m.get(k) : null },
    async list (prefix = '') { const o = {}; for (const [k, v] of m) if (k.startsWith(prefix)) o[k] = v; return o }
  }
}

function clientRpc (swarm, serverKeyHex) {
  return new Promise((resolve) => {
    swarm.on('connection', (conn) => resolve(new ProtomuxRPC(conn, { protocol: DEVICE_PROTOCOL, valueEncoding: c.json })))
    swarm.join(b4a.from(serverKeyHex, 'hex'), { client: true, server: false })
  })
}

test('device control surface: gated apply-write + observable get-state over swarm', async (t) => {
  const testnet = await createTestnet(3)
  const device = new SwarmDevice(memKV(), { did: DID, privateKey: crypto.randomBytes(32), self: 'deviceA' })

  const serverSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const authedClient = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const strangerClient = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const authedPub = b4a.toString(authedClient.keyPair.publicKey, 'hex')

  // canWrite allowlist = only the authed client's Noise pubkey (secure default is null elsewhere).
  const serving = serveDevice(device, serverSwarm, { canWrite: (hex) => hex === authedPub })
  await serving.ready()
  t.teardown(async () => {
    serving.close()
    await serverSwarm.destroy(); await authedClient.destroy(); await strangerClient.destroy()
    await testnet.destroy()
  })

  const rpc = await clientRpc(authedClient, serving.serverKey)

  // get-state before any write — observable, empty.
  const s0 = await rpc.request('get-state', {})
  t.is(s0.did, DID, 'get-state reports the DID')
  t.is(s0.recordCount, 0, 'no records before write')

  // authorized apply-write lands the intent.
  const rec = { $type: POST, text: 'via control surface', createdAt: '2026-07-04T00:00:00.000Z' }
  const w = await rpc.request('apply-write', { action: 'create', collection: POST, rkey: '3ctrl', record: rec })
  t.is(w.recordCount, 1, 'authorized write landed')

  // get-state now reflects it (keys only).
  const s1 = await rpc.request('get-state', {})
  t.is(s1.recordCount, 1, 'record count updated')
  t.alike(s1.records[0], { collection: POST, rkey: '3ctrl' }, 'record observable via get-state')

  // a non-allowlisted client is refused write but may still read.
  const rpc2 = await clientRpc(strangerClient, serving.serverKey)
  const s2 = await rpc2.request('get-state', {})
  t.is(s2.recordCount, 1, 'stranger may read state')
  // gating: the stranger's write is refused (the RPC rejects; protomux does not forward the
  // server message, so assert by behavior) and does NOT land — record count is unchanged.
  await t.exception(() => rpc2.request('apply-write', { action: 'create', collection: POST, rkey: 'x', record: rec }), 'stranger write rejected')
  const s3 = await rpc2.request('get-state', {})
  t.is(s3.recordCount, 1, 'rejected write did not land')

  // pointer-diag is a READ, but it is gated with the WRITES rather than published with get-state:
  // it hashes every record file, so an ungated responder would hand any peer that knows the DID a free
  // CPU-exhaustion trigger (the same reason reindex-overlay stays UDS-only). Gating it on the existing
  // trusted-key allowlist is what makes `hyper-proto-peer diag --remote` able to ask "do your bytes match
  // your pointers?" without an ssh session, while keeping the cost behind a key.
  const pd = await rpc.request('pointer-diag', {})
  t.ok(pd, 'an ALLOWLISTED caller reaches pointer-diag remotely')
  t.is(pd.skipped, 'no-record-store', 'and gets the device\'s real answer (this device has no record store)')
  await t.exception(() => rpc2.request('pointer-diag', {}), 'a stranger is refused pointer-diag — it is not a public read')
})