/**
* rpc-audit-log: every dispatch, on every transport, lands in an append-only
* JSONL record with the caller's resolved userId.
*
* WHY: the daemon's control plane kept no record of who called what. On
* 2026-07-29 three repos were pulled onto a daemon via join-repo over
* /api/rpc; the method name lives in the (unlogged) request body, so the
* calls were only reconstructible by correlating an upstream Apache log
* against daemon-side side-effects, and several remained unattributed.
*/
import test from 'brittle'
import { mkdtemp, rm, readFile } from '#node:fs/promises'
import { tmpdir } from '#node:os'
import { join } from '#node:path'
import { installRpcAuditLog, RpcAuditLog } from '../lib/services/rpc-audit-log.js'
function fakeSvc (handlers) {
return {
async dispatch (method, params, ctx) {
const h = handlers[method]
if (!h) throw new Error('unknown method: ' + method)
return h(ctx, params)
}
}
}
const registry = {
async getUserByIdentity (pk) {
return pk === 'pubkey-aaa' ? { userId: 'user-abc' } : null
}
}
async function readLines (path) {
const raw = await readFile(path, 'utf8')
return raw.trim().split('\n').filter(Boolean).map((l) => JSON.parse(l))
}
test('rpc-audit-log: records method, transport and resolved userId', async (t) => {
const root = await mkdtemp(join(tmpdir(), 'rpcaudit-'))
t.teardown(() => rm(root, { recursive: true, force: true }))
const path = join(root, 'rpc-audit.jsonl')
const svc = fakeSvc({ 'join-repo': async () => ({ name: 'demo' }) })
const audit = installRpcAuditLog(svc, { path, registry, skip: [] })
const ctx = {
transport: 'http',
identity: { publicKey: 'pubkey-aaa' },
origin: { remoteAddress: '179.43.189.87', userAgent: null }
}
const out = await svc.dispatch('join-repo', { invitation: 'roompair1:x' }, ctx)
t.is(out.name, 'demo', 'dispatch return value is passed through untouched')
await audit.close()
const lines = await readLines(path)
t.is(lines.length, 1, 'one entry recorded')
t.is(lines[0].method, 'join-repo', 'method recorded')
t.is(lines[0].transport, 'http', 'transport recorded')
t.is(lines[0].userId, 'user-abc', 'caller resolved to their userId')
t.is(lines[0].remote, '179.43.189.87', 'source address recorded')
t.is(lines[0].ok, true, 'success recorded')
})
test('rpc-audit-log: records anonymous callers and failures', async (t) => {
const root = await mkdtemp(join(tmpdir(), 'rpcaudit-'))
t.teardown(() => rm(root, { recursive: true, force: true }))
const path = join(root, 'rpc-audit.jsonl')
const svc = fakeSvc({ boom: async () => { throw new Error('nope') } })
const audit = installRpcAuditLog(svc, { path, registry, skip: [] })
// Anonymous /rpc/<method> call — identity null, as hyper-svc builds it.
await t.exception(
() => svc.dispatch('boom', {}, { transport: 'http', identity: null }),
'the wrapper rethrows the handler error unchanged'
)
await audit.close()
const lines = await readLines(path)
t.is(lines.length, 1, 'failed call still recorded')
t.is(lines[0].userId, null, 'anonymous caller logged as null, not dropped')
t.is(lines[0].ok, false, 'failure recorded')
t.is(lines[0].error, 'nope', 'error message recorded')
})
test('rpc-audit-log: swarm callers are identified, not logged as anonymous', async (t) => {
const root = await mkdtemp(join(tmpdir(), 'rpcaudit-'))
t.teardown(() => rm(root, { recursive: true, force: true }))
const path = join(root, 'rpc-audit.jsonl')
const svc = fakeSvc({ 'list-orgs': async () => [] })
const audit = installRpcAuditLog(svc, { path, registry, skip: [] })
// The shape hyper-svc's swarm-rpc transport actually builds: non-null
// identity, NO publicKey, trust:'network'.
await svc.dispatch('list-orgs', {}, {
transport: 'swarm-rpc',
identity: { type: 'swarm', remotePublicKey: 'abcdef0123456789aa', trust: 'network' }
})
await audit.close()
const lines = await readLines(path)
t.is(lines[0].identity, 'abcdef0123456789', 'swarm peer key recorded')
t.is(lines[0].trust, 'network', 'network trust is distinguishable from an authenticated user')
t.is(lines[0].userId, null, 'a swarm peer maps to no user')
})
test('rpc-audit-log: skip-list keeps high-frequency reads out', async (t) => {
const root = await mkdtemp(join(tmpdir(), 'rpcaudit-'))
t.teardown(() => rm(root, { recursive: true, force: true }))
const path = join(root, 'rpc-audit.jsonl')
const svc = fakeSvc({ status: async () => ({}), 'create-access-token': async () => ({}) })
const audit = installRpcAuditLog(svc, { path, registry, skip: ['status'] })
await svc.dispatch('status', {}, { transport: 'uds' })
await svc.dispatch('create-access-token', {}, { transport: 'http', identity: { publicKey: 'pubkey-aaa' } })
await audit.close()
const lines = await readLines(path)
t.is(lines.length, 1, 'only the non-skipped call recorded')
t.is(lines[0].method, 'create-access-token', 'the mutation is what survived')
})
test('rpc-audit-log: a logging failure never breaks dispatch', async (t) => {
// Unwritable path: /dev/null is a file, so mkdir + append below it fail with
// ENOTDIR. Every write throws internally and must be swallowed.
const svc = fakeSvc({ ping: async () => 'pong' })
const audit = installRpcAuditLog(svc, { path: '/dev/null/nope/audit.jsonl', registry, skip: [] })
t.teardown(() => audit.close())
const out = await svc.dispatch('ping', {}, { transport: 'uds' })
t.is(out, 'pong', 'dispatch still returns normally when logging cannot write')
await audit.flush()
t.pass('flushing an unwritable log does not throw')
})
test('rpc-audit-log: userId cache does not grow without bound', async (t) => {
const audit = new RpcAuditLog({ path: join(tmpdir(), 'unused.jsonl'), registry: { async getUserByIdentity () { return { userId: 'u' } } } })
for (let i = 0; i < 600; i++) await audit._userIdFor('pk-' + i)
t.ok(audit._userIds.size <= 500, 'cache is bounded (' + audit._userIds.size + ')')
})